Pectra changes. Consolidate + new reward calculation (#238)

Author: Marketen

api/api.go

@@ -365,6 +365,29 @@ func (m *ApiService) handleMemoryStatistics(w http.ResponseWriter, req *http.Req
 		avgBlockRewardWei = big.NewInt(0).Div(totalRewardsSentWei, big.NewInt(0).SetUint64(totalOkPoolProposalBlocks))
 	}
 
+	// Total and average effective balance
+	totalEffectiveBalance := uint64(0)
+	effectiveBalances := m.Onchain.Validators()
+
+	for _, validator := range m.oracle.State().Validators {
+		if validator.ValidatorStatus == oracle.Active ||
+			validator.ValidatorStatus == oracle.YellowCard ||
+			validator.ValidatorStatus == oracle.RedCard {
+
+			beaconState, found := effectiveBalances[phase0.ValidatorIndex(validator.ValidatorIndex)]
+			if !found {
+				log.Warn("could not find validator in beacon state: ", validator.ValidatorIndex)
+				continue
+			}
+			totalEffectiveBalance += uint64(beaconState.Validator.EffectiveBalance)
+		}
+	}
+
+	avgEffectiveBalance := uint64(0)
+	if totalSubscribed > 0 {
+		avgEffectiveBalance = totalEffectiveBalance / totalSubscribed
+	}
+
 	m.respondOK(w, httpOkMemoryStatistics{
 		TotalSubscribed:              totalSubscribed,
 		TotalActive:                  totalActive,
@@ -384,6 +407,8 @@ func (m *ApiService) handleMemoryStatistics(w http.ResponseWriter, req *http.Req
 		TotalProposedBlocks:          totalProposedBlocks,
 		TotalMissedBlocks:            uint64(len(m.oracle.State().MissedBlocks)),
 		TotalWrongFeeBlocks:          uint64(len(m.oracle.State().WrongFeeBlocks)),
+		TotalEffectiveBalanceGwei:    fmt.Sprintf("%d", totalEffectiveBalance),
+		AvgEffectiveBalanceGwei:      fmt.Sprintf("%d", avgEffectiveBalance),
 	})
 }
 
@@ -515,15 +540,17 @@ func (m *ApiService) handleMemoryValidators(w http.ResponseWriter, req *http.Req
 			continue
 		}
 		validatorsResp = append(validatorsResp, httpOkValidatorInfo{
-			ValidatorStatus:       v.ValidatorStatus.String(),
-			BeaconValidatorStatus: beaconState.Status.String(),
-			AccumulatedRewardsWei: v.AccumulatedRewardsWei.String(),
-			PendingRewardsWei:     v.PendingRewardsWei.String(),
-			CollateralWei:         v.CollateralWei.String(),
-			WithdrawalAddress:     v.WithdrawalAddress,
-			ValidatorIndex:        v.ValidatorIndex,
-			ValidatorKey:          v.ValidatorKey,
-			SubscriptionType:      v.SubscriptionType.String(),
+			ValidatorStatus:                     v.ValidatorStatus.String(),
+			BeaconValidatorBalanceGwei:          fmt.Sprintf("%d", beaconState.Balance),
+			BeaconValidatorEffectiveBalanceGwei: fmt.Sprintf("%d", beaconState.Validator.EffectiveBalance),
+			BeaconValidatorStatus:               beaconState.Status.String(),
+			AccumulatedRewardsWei:               v.AccumulatedRewardsWei.String(),
+			PendingRewardsWei:                   v.PendingRewardsWei.String(),
+			CollateralWei:                       v.CollateralWei.String(),
+			WithdrawalAddress:                   v.WithdrawalAddress,
+			ValidatorIndex:                      v.ValidatorIndex,
+			ValidatorKey:                        v.ValidatorKey,
+			SubscriptionType:                    v.SubscriptionType.String(),
 		})
 	}
 
@@ -601,15 +628,17 @@ func (m *ApiService) handleMemoryValidatorsByIndex(w http.ResponseWriter, req *h
 				continue
 			}
 			foundValidator := httpOkValidatorInfo{
-				ValidatorStatus:       validator.ValidatorStatus.String(),
-				BeaconValidatorStatus: beaconState.Status.String(),
-				AccumulatedRewardsWei: validator.AccumulatedRewardsWei.String(),
-				PendingRewardsWei:     validator.PendingRewardsWei.String(),
-				CollateralWei:         validator.CollateralWei.String(),
-				WithdrawalAddress:     validator.WithdrawalAddress,
-				ValidatorIndex:        validator.ValidatorIndex,
-				ValidatorKey:          validator.ValidatorKey,
-				SubscriptionType:      validator.SubscriptionType.String(),
+				ValidatorStatus:                     validator.ValidatorStatus.String(),
+				BeaconValidatorBalanceGwei:          fmt.Sprintf("%d", beaconState.Balance),
+				BeaconValidatorEffectiveBalanceGwei: fmt.Sprintf("%d", beaconState.Validator.EffectiveBalance),
+				BeaconValidatorStatus:               beaconState.Status.String(),
+				AccumulatedRewardsWei:               validator.AccumulatedRewardsWei.String(),
+				PendingRewardsWei:                   validator.PendingRewardsWei.String(),
+				CollateralWei:                       validator.CollateralWei.String(),
+				WithdrawalAddress:                   validator.WithdrawalAddress,
+				ValidatorIndex:                      validator.ValidatorIndex,
+				ValidatorKey:                        validator.ValidatorKey,
+				SubscriptionType:                    validator.SubscriptionType.String(),
 			}
 			foundValidators = append(foundValidators, foundValidator)
 		} else {
@@ -754,15 +783,17 @@ func (m *ApiService) handleMemoryValidatorsByWithdrawal(w http.ResponseWriter, r
 			continue
 		}
 		validatorsResp = append(validatorsResp, httpOkValidatorInfo{
-			ValidatorStatus:       v.ValidatorStatus.String(),
-			BeaconValidatorStatus: beaconState.Status.String(),
-			AccumulatedRewardsWei: v.AccumulatedRewardsWei.String(),
-			PendingRewardsWei:     v.PendingRewardsWei.String(),
-			CollateralWei:         v.CollateralWei.String(),
-			WithdrawalAddress:     v.WithdrawalAddress,
-			ValidatorIndex:        v.ValidatorIndex,
-			ValidatorKey:          v.ValidatorKey,
-			SubscriptionType:      v.SubscriptionType.String(),
+			ValidatorStatus:                     v.ValidatorStatus.String(),
+			BeaconValidatorBalanceGwei:          fmt.Sprintf("%d", beaconState.Balance),
+			BeaconValidatorEffectiveBalanceGwei: fmt.Sprintf("%d", beaconState.Validator.EffectiveBalance),
+			BeaconValidatorStatus:               beaconState.Status.String(),
+			AccumulatedRewardsWei:               v.AccumulatedRewardsWei.String(),
+			PendingRewardsWei:                   v.PendingRewardsWei.String(),
+			CollateralWei:                       v.CollateralWei.String(),
+			WithdrawalAddress:                   v.WithdrawalAddress,
+			ValidatorIndex:                      v.ValidatorIndex,
+			ValidatorKey:                        v.ValidatorKey,
+			SubscriptionType:                    v.SubscriptionType.String(),
 		})
 	}
 	m.respondOK(w, validatorsResp)

api/types.go

@@ -79,6 +79,8 @@ type httpOkMemoryStatistics struct {
 	TotalProposedBlocks          uint64 `json:"total_proposed_blocks"`
 	TotalMissedBlocks            uint64 `json:"total_missed_blocks"`
 	TotalWrongFeeBlocks          uint64 `json:"total_wrongfee_blocks"`
+	TotalEffectiveBalanceGwei    string `json:"total_effective_balance_gwei"`
+	AvgEffectiveBalanceGwei      string `json:"avg_effective_balance_gwei"`
 }
 
 type httpOkProofs struct {
@@ -130,15 +132,17 @@ type httpOkBlock struct {
 }
 
 type httpOkValidatorInfo struct {
-	ValidatorStatus       string `json:"status"`
-	BeaconValidatorStatus string `json:"beacon_status"`
-	AccumulatedRewardsWei string `json:"accumulated_rewards_wei"`
-	PendingRewardsWei     string `json:"pending_rewards_wei"`
-	CollateralWei         string `json:"collateral_wei"`
-	WithdrawalAddress     string `json:"withdrawal_address"`
-	ValidatorIndex        uint64 `json:"validator_index"`
-	ValidatorKey          string `json:"validator_key"`
-	SubscriptionType      string `json:"subscription_type"`
+	ValidatorStatus                     string `json:"status"`
+	BeaconValidatorStatus               string `json:"beacon_status"`
+	BeaconValidatorBalanceGwei          string `json:"beacon_balance_gwei"`
+	BeaconValidatorEffectiveBalanceGwei string `json:"beacon_effective_balance_gwei"`
+	AccumulatedRewardsWei               string `json:"accumulated_rewards_wei"`
+	PendingRewardsWei                   string `json:"pending_rewards_wei"`
+	CollateralWei                       string `json:"collateral_wei"`
+	WithdrawalAddress                   string `json:"withdrawal_address"`
+	ValidatorIndex                      uint64 `json:"validator_index"`
+	ValidatorKey                        string `json:"validator_key"`
+	SubscriptionType                    string `json:"subscription_type"`
 }
 
 type httpOkValidatorsByIndex struct {

main.go

@@ -129,6 +129,7 @@ func main() {
 	// Create the oracle instance
 	oracleInstance := oracle.NewOracle(cfg)
 	oracleInstance.SetGetSetOfValidatorsFunc(onchain.GetSetOfValidators)
+	oracleInstance.GetPendingConsolidationsFunc(onchain.GetPendingConsolidations)
 
 	// If checkpoint sync url is provided, load state from it
 	if cliCfg.CheckPointSyncUrl != "" {
@@ -309,7 +310,7 @@ func mainLoop(oracleInstance *oracle.Oracle, onchain *oracle.Onchain, cfg *oracl
 			continue
 		}
 
-		// Every X slots we update the onchain validators and cleanup any stranded oracle validators
+		// Every X slots we update the onchain validators
 		if oracleInstance.State().LatestProcessedSlot%UpdateValidatorsIntervalSlots == 0 {
 			onchain.RefreshBeaconValidators()
 		}

oracle/onchain.go

@@ -4,8 +4,11 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
+	"io"
 	"math/big"
+	normalHttp "net/http"
 	"strconv"
 	"strings"
 	"time"
@@ -40,6 +43,12 @@ var GoerliChainId = uint64(5)
 var HoleskyChainId = uint64(17000)
 var HoodiChainId = uint64(560048)
 
+// Network names
+var Mainnet = "mainnet"
+var Goerli = "goerli"
+var Holesky = "holesky"
+var Hoodi = "hoodi"
+
 // This file provides different functions to access the blockchain state from both consensus and
 // execution layer and modifying the its state via smart contract calls.
 type EpochDuties struct {
@@ -1063,13 +1072,13 @@ func (onchain *Onchain) GetConfigFromContract(
 
 	network := ""
 	if depositContract.Data.ChainID == MainnetChainId {
-		network = "mainnet"
+		network = Mainnet
 	} else if depositContract.Data.ChainID == GoerliChainId {
-		network = "goerli"
+		network = Goerli
 	} else if depositContract.Data.ChainID == HoleskyChainId {
-		network = "holesky"
+		network = Holesky
 	} else if depositContract.Data.ChainID == HoodiChainId {
-		network = "hoodi"
+		network = Hoodi
 	} else {
 		log.Fatal("ChainID not supported: ", depositContract.Data.ChainID)
 	}
@@ -1763,3 +1772,62 @@ func (o *Onchain) GetRetryOpts(opts []retry.Option) []retry.Option {
 		return opts
 	}
 }
+
+// TODO: Manual implementation of pending consolidations beacon API call
+// Remove once attestantio eth2 library is updated to support it
+type PendingConsolidation struct {
+	SourceIndex phase0.ValidatorIndex `json:"source_index"`
+	TargetIndex phase0.ValidatorIndex `json:"target_index"`
+}
+
+type PendingConsolidationsResponse struct {
+	Version             string                 `json:"version"`
+	ExecutionOptimistic bool                   `json:"execution_optimistic"`
+	Finalized           bool                   `json:"finalized"`
+	Data                []PendingConsolidation `json:"data"`
+}
+
+func (o *Onchain) GetPendingConsolidations(stateID string, opts ...retry.Option) (*PendingConsolidationsResponse, error) {
+	var (
+		consolidations *PendingConsolidationsResponse
+		err            error
+	)
+
+	err = retry.Do(func() error {
+		url := fmt.Sprintf("%s/eth/v1/beacon/states/%s/pending_consolidations", o.ConsensusClient.Address(), stateID)
+
+		req, reqErr := normalHttp.NewRequest(normalHttp.MethodGet, url, nil)
+		if reqErr != nil {
+			return fmt.Errorf("creating request failed: %w", reqErr)
+		}
+		req.Header.Set("Accept", "application/json")
+
+		resp, httpErr := normalHttp.DefaultClient.Do(req)
+		if httpErr != nil {
+			log.Warn("Failed attempt to call pending consolidations endpoint: ", httpErr.Error(), " Retrying...")
+			return fmt.Errorf("HTTP request failed: %w", httpErr)
+		}
+		defer resp.Body.Close()
+
+		if resp.StatusCode != normalHttp.StatusOK {
+			bodyBytes, _ := io.ReadAll(resp.Body)
+			log.Warn("Received non-200 response: ", resp.StatusCode, " Body: ", string(bodyBytes))
+			return fmt.Errorf("non-200 response: %d - %s", resp.StatusCode, string(bodyBytes))
+		}
+
+		var result PendingConsolidationsResponse
+		if decodeErr := json.NewDecoder(resp.Body).Decode(&result); decodeErr != nil {
+			log.Warn("Failed attempt to decode pending consolidations response: ", decodeErr.Error(), " Retrying...")
+			return fmt.Errorf("decoding response failed: %w", decodeErr)
+		}
+
+		consolidations = &result
+		return nil
+	}, o.GetRetryOpts(opts)...)
+
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch pending consolidations: %w", err)
+	}
+
+	return consolidations, nil
+}

oracle/onchain_test.go

@@ -470,6 +470,27 @@ func Test_EndToEnd_Mainnet(t *testing.T) {
 	}
 }
 
+func Test_GetPendingConsolidations(t *testing.T) {
+	// Note it works only for hoodi
+	t.Skip("Skipping test")
+
+	var cfgOnchain = &config.CliConfig{
+		ConsensusEndpoint: "http://127.0.0.1:3500",
+		ExecutionEndpoint: "http://127.0.0.1:8545",
+		PoolAddress:       "0xAdFb8D27671F14f297eE94135e266aAFf8752e35",
+	}
+
+	onchain, err := NewOnchain(cfgOnchain, nil)
+	require.NoError(t, err)
+
+	consolidations, err := onchain.GetPendingConsolidations("100000")
+	require.NoError(t, err)
+
+	for _, c := range consolidations.Data {
+		fmt.Println(c.SourceIndex, c.TargetIndex)
+	}
+}
+
 func Test_Mainnet_BeaverIssue(t *testing.T) {
 	t.Skip("Skipping test")