Merge pull request #790 from lidofinance/feat/validator-stages

feat: use lazy oracle to check validator stages

Author: tamtamchik

README.md

@@ -222,7 +222,8 @@ In manual mode all sleeps are disabled and `ALLOW_REPORTING_IN_BUNKER_MODE` is T
 | `CACHE_PATH`                                           | Directory to store cache for CSM module                                                                                                                                  | False    | `.`                     |
 | `OPSGENIE_API_KEY`                                     | OpsGenie API key for authentication with the OpsGenie API. Used to send alerts from lido-oracle health-checks.                                                           | False    | `<api-key>`             |
 | `OPSGENIE_API_URL`                                     | Base URL for the OpsGenie API.                                                                                                                                           | False    | `http://localhost:8080` |
-| `VAULT_PAGINATION_LIMIT`                               | The limit for getting staking vaults with pagination. Default 1000                                                                                                       | False    | `http://localhost:8080` |
+| `VAULT_PAGINATION_LIMIT`                               | The limit for getting staking vaults with pagination. Default 1000                                                                                                       | False    | `1000`                  |
+| `VAULT_VALIDATOR_STAGES_BATCH_SIZE`                    | The limit for getting validators stages in one request. Default 100                                                                                                      | False    | `100`                   |
 
 ### Mainnet variables
 > LIDO_LOCATOR_ADDRESS=0xC1d0b3DE6792Bf6b4b37EccdcC24e45978Cfd2Eb

assets/LazyOracle.json

@@ -18,7 +18,7 @@ license = "GPL 3.0"
 
 [tool.poetry.dependencies]
 python = "^3.12"
-prometheus-client = "0.21.1"
+prometheus-client = "^0.21.1"
 timeout-decorator = "^0.5.0"
 pytest = "^7.2.1"
 pytest-xdist = "^3.2.1"

pyproject.toml

@@ -14,10 +14,10 @@
 # https://github.com/ethereum/consensus-specs/blob/dev/specs/phase0/beacon-chain.md#gwei-values
 EFFECTIVE_BALANCE_INCREMENT = Gwei(2**0 * 10**9)
 MAX_EFFECTIVE_BALANCE = Gwei(32 * 10**9)
+MIN_DEPOSIT_AMOUNT = Gwei(2**0 * 10**9)
 # https://github.com/ethereum/consensus-specs/blob/dev/specs/electra/beacon-chain.md#gwei-values
 MAX_EFFECTIVE_BALANCE_ELECTRA = Gwei(2**11 * 10**9)
 MIN_ACTIVATION_BALANCE = Gwei(2**5 * 10**9)
-PDG_ACTIVATION_DEPOSIT = Gwei(31 * 10**9)
 # https://github.com/ethereum/consensus-specs/blob/dev/specs/capella/beacon-chain.md#execution
 MAX_WITHDRAWALS_PER_PAYLOAD = 2**4
 # https://github.com/ethereum/consensus-specs/blob/dev/specs/phase0/beacon-chain.md#withdrawal-prefixes

src/constants.py

@@ -243,12 +243,14 @@ def _get_consensus_lido_state(self, blockstamp: ReferenceBlockStamp) -> tuple[Va
         total_lido_balance = lido_validators_state_balance = sum((validator.balance for validator in lido_validators), Gwei(0))
         logger.info({
             'msg': 'Calculate Lido validators state balance (in Gwei)',
-            'value': lido_validators_state_balance
+            'value': lido_validators_state_balance,
         })
 
         return ValidatorsCount(len(lido_validators)), ValidatorsBalance(Gwei(total_lido_balance))
 
-    def _get_finalization_data(self, blockstamp: ReferenceBlockStamp) -> tuple[FinalizationBatches, FinalizationShareRate]:
+    def _get_finalization_data(
+        self, blockstamp: ReferenceBlockStamp
+    ) -> tuple[FinalizationBatches, FinalizationShareRate]:
         simulation = self.simulate_full_rebase(blockstamp)
         chain_config = self.get_chain_config(blockstamp)
         frame_config = self.get_frame_config(blockstamp)
@@ -412,7 +414,7 @@ def _handle_vaults_report(self, blockstamp: ReferenceBlockStamp) -> VaultsReport
 
         vaults = self.staking_vaults.get_vaults(blockstamp.block_hash)
         if len(vaults) == 0:
-            return b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', ''
+            return ZERO_HASH, ''
 
         current_frame = self.get_frame_number_by_slot(blockstamp)
         validators = self.w3.cc.get_validators(blockstamp)
@@ -421,6 +423,13 @@ def _handle_vaults_report(self, blockstamp: ReferenceBlockStamp) -> VaultsReport
         frame_config = self.get_frame_config(blockstamp)
         simulation = self.simulate_full_rebase(blockstamp)
 
+        vaults_total_values = self.staking_vaults.get_vaults_total_values(
+            vaults=vaults,
+            validators=validators,
+            pending_deposits=pending_deposits,
+            block_identifier=blockstamp.block_hash
+        )
+
         core_apr_ratio = calculate_gross_core_apr(
             pre_total_ether=simulation.pre_total_pooled_ether,
             pre_total_shares=simulation.pre_total_shares,
@@ -430,12 +439,6 @@ def _handle_vaults_report(self, blockstamp: ReferenceBlockStamp) -> VaultsReport
             time_elapsed_seconds=self._get_time_elapsed_seconds_from_prev_report(blockstamp),
         )
 
-        vaults_total_values = self.staking_vaults.get_vaults_total_values(
-            vaults=vaults,
-            validators=validators,
-            pending_deposits=pending_deposits,
-        )
-
         latest_onchain_ipfs_report_data = self.staking_vaults.get_latest_onchain_ipfs_report_data(blockstamp.block_hash)
         vaults_fees = self.staking_vaults.get_vaults_fees(
             blockstamp=blockstamp,

src/modules/accounting/accounting.py

@@ -1,10 +1,11 @@
 from dataclasses import dataclass
+from typing import Any, Dict, Union
 
 from eth_typing import BlockNumber
 from hexbytes import HexBytes
-from typing import Any, Dict, Union
 from web3.types import EventData
 
+
 @dataclass(frozen=True)
 class EventBase:
     event: str
@@ -144,6 +145,7 @@ def from_log(cls, log: EventData) -> "VaultConnectedEvent":
             **cls._extract_common(log),
         )
 
+
 @dataclass(frozen=True)
 class BadDebtWrittenOffToBeInternalizedEvent(EventBase):
     vault: str
@@ -158,14 +160,15 @@ def from_log(cls, log: EventData) -> "BadDebtWrittenOffToBeInternalizedEvent":
             **cls._extract_common(log),
         )
 
+
 VaultEventType = Union[
     MintedSharesOnVaultEvent,
     BurnedSharesOnVaultEvent,
     VaultFeesUpdatedEvent,
     VaultRebalancedEvent,
     BadDebtSocializedEvent,
     BadDebtWrittenOffToBeInternalizedEvent,
-    VaultConnectedEvent
+    VaultConnectedEvent,
 ]
 
 

src/modules/accounting/events.py

@@ -132,7 +132,7 @@ def from_response(cls, **kwargs) -> Self:
 class BatchState:
     remaining_eth_budget: int
     finished: bool
-    batches: tuple[int, ...]
+    batches: list[int]
     batches_length: int
 
     def as_tuple(self):
@@ -231,7 +231,7 @@ class OnChainIpfsVaultReportData(Nested, FromResponse):
 @dataclass
 class VaultInfo(Nested, FromResponse):
     vault: ChecksumAddress
-    aggregate_balance: Wei
+    aggregated_balance: Wei
     in_out_delta: Wei
     withdrawal_credentials: str
     liability_shares: Shares
@@ -417,3 +417,10 @@ def total(self) -> Gwei:
     @property
     def max(self) -> Gwei:
         return Gwei(max((deposit.amount for deposit in self.pending_deposits), default=0))
+
+class ValidatorStage(Enum):
+    NONE = 0
+    PREDEPOSITED = 1
+    PROVEN = 2
+    ACTIVATED = 3
+    COMPENSATED = 4

src/modules/accounting/types.py

@@ -250,7 +250,7 @@ def get_state_view_no_cache(self, blockstamp: BlockStamp) -> BeaconStateView:
     def get_pending_deposits(self, blockstamp: BlockStamp) -> list[PendingDeposit]:
         return self.get_state_view(blockstamp).pending_deposits
 
-    def get_validator_state(self, state_id: SlotNumber | BlockRoot, validator_id: int) -> Validator:
+    def get_validator_state(self, state_id: SlotNumber, validator_id: int) -> Validator:
         """Spec: https://ethereum.github.io/beacon-APIs/#/Beacon/getStateValidator"""
         data, _ = self._get(
             self.API_GET_VALIDATOR,

src/providers/consensus/client.py

@@ -2,7 +2,10 @@
 
 from web3.types import BlockIdentifier
 
-from src.modules.accounting.types import ReportSimulationPayload, ReportSimulationResults
+from src.modules.accounting.types import (
+    ReportSimulationPayload,
+    ReportSimulationResults,
+)
 from src.providers.execution.base_interface import ContractInterface
 
 logger = logging.getLogger(__name__)
@@ -23,9 +26,7 @@ def simulate_oracle_report(
         plugging the returned values to the following formula: `_simulatedShareRate = (postTotalPooledEther * 1e27) / postTotalShares`
         """
 
-        response = self.functions.simulateOracleReport(payload.as_tuple()).call(
-            block_identifier=block_identifier
-        )
+        response = self.functions.simulateOracleReport(payload.as_tuple()).call(block_identifier=block_identifier)
 
         response = ReportSimulationResults(*response)
 

src/providers/execution/contracts/accounting.py

@@ -1,18 +1,23 @@
 import logging
 
+from hexbytes import HexBytes
 from web3 import Web3
 from web3.types import BlockIdentifier
 
 from src import variables
-from src.modules.accounting.types import OnChainIpfsVaultReportData, VaultInfo
+from src.modules.accounting.types import (
+    OnChainIpfsVaultReportData,
+    ValidatorStage,
+    VaultInfo,
+)
 from src.providers.execution.base_interface import ContractInterface
 from src.utils.abi import named_tuple_to_dataclass
 from src.utils.cache import global_lru_cache as lru_cache
 
 logger = logging.getLogger(__name__)
 
 
-class VaultsLazyOracleContract(ContractInterface):
+class LazyOracleContract(ContractInterface):
     abi_path = './assets/LazyOracle.json'
 
     @lru_cache(maxsize=1)
@@ -22,28 +27,24 @@ def get_vaults_count(self, block_identifier: BlockIdentifier = 'latest') -> int:
         """
         response = self.functions.vaultsCount.call(block_identifier=block_identifier)
 
-        logger.info(
-            {
-                'msg': 'Call `vaultsCount().',
-                'value': response,
-                'block_identifier': repr(block_identifier),
-                'to': self.address,
-            }
-        )
+        logger.info({
+            'msg': 'Call `vaultsCount().',
+            'value': response,
+            'block_identifier': repr(block_identifier),
+            'to': self.address,
+        })
 
         return response
 
     def get_latest_report_data(self, block_identifier: BlockIdentifier = 'latest') -> OnChainIpfsVaultReportData:
         response = self.functions.latestReportData.call(block_identifier=block_identifier)
 
-        logger.info(
-            {
-                'msg': 'Call `latestReportData()`.',
-                'value': response,
-                'block_identifier': repr(block_identifier),
-                'to': self.address,
-            }
-        )
+        logger.info({
+            'msg': 'Call `latestReportData()`.',
+            'value': response,
+            'block_identifier': repr(block_identifier),
+            'to': self.address,
+        })
 
         response = named_tuple_to_dataclass(response, OnChainIpfsVaultReportData)
         return response
@@ -58,7 +59,7 @@ def get_vaults(self, offset: int, limit: int, block_identifier: BlockIdentifier
         for vault in response:
             out.append(VaultInfo(
                 vault=vault.vault,
-                aggregate_balance=vault.aggregateBalance,
+                aggregated_balance=vault.aggregatedBalance,
                 in_out_delta=vault.inOutDelta,
                 withdrawal_credentials=Web3.to_hex(vault.withdrawalCredentials),
                 liability_shares=vault.liabilityShares,
@@ -73,14 +74,12 @@ def get_vaults(self, offset: int, limit: int, block_identifier: BlockIdentifier
                 pending_disconnect=vault.pendingDisconnect,
             ))
 
-        logger.info(
-            {
-                'msg': f'Call `batchVaultsInfo({offset}, {limit}).',
-                'value': response,
-                'block_identifier': repr(block_identifier),
-                'to': self.address,
-            }
-        )
+        logger.info({
+            'msg': f'Call `batchVaultsInfo({offset}, {limit}).',
+            'value': response,
+            'block_identifier': repr(block_identifier),
+            'to': self.address,
+        })
 
         return out
 
@@ -96,10 +95,40 @@ def get_all_vaults(self, block_identifier: BlockIdentifier = 'latest') -> list[V
             return []
 
         while offset < total_count:
-            batch = self.get_vaults(block_identifier=block_identifier, offset=offset, limit=variables.VAULT_PAGINATION_LIMIT)
+            batch = self.get_vaults(
+                block_identifier=block_identifier, offset=offset, limit=variables.VAULT_PAGINATION_LIMIT
+            )
             if not batch:
                 break
             vaults.extend(batch)
             offset += variables.VAULT_PAGINATION_LIMIT
 
         return vaults
+
+    def get_validator_stages(
+        self,
+        pubkeys: list[str],
+        batch_size: int = variables.VAULT_VALIDATOR_STAGES_BATCH_SIZE,
+        block_identifier: BlockIdentifier = 'latest'
+    ) -> dict[str, ValidatorStage]:
+        """
+        Fetch validator stages for a list of pubkeys, batching requests for efficiency.
+        """
+        out: dict[str, ValidatorStage] = {}
+
+        for i in range(0, len(pubkeys), batch_size):
+            batch = list(map(HexBytes, pubkeys[i : i + batch_size]))
+            response = self.functions.batchValidatorStages(batch).call(block_identifier=block_identifier)
+
+            logger.debug({
+                'msg': 'Call `batchValidatorStages()`.',
+                'count': len(batch),
+                'block_identifier': repr(block_identifier),
+                'to': self.address,
+            })
+
+            out.update({
+                str(pk): ValidatorStage(stage) for pk, stage in zip(batch, response)
+            })
+
+        return out