feat: CSM contracts as a separate module

assets/CSModule.json

@@ -0,0 +1 @@
+[{"type":"event","name":"StuckSigningKeysCountChanged","inputs":[{"name":"nodeOperatorId","type":"uint256","indexed":true,"internalType":"uint256"},{"name":"stuckKeysCount","type":"uint256","indexed":false,"internalType":"uint256"}],"anonymous":false}]

fixtures/common/contracts.json

@@ -118,21 +118,5 @@
       "id": 5,
       "result": "0x000000000000000000000000655617d5fb44649e979e41229cbae9e9b63c0a2a"
     }
-  },
-  {
-    "method": "eth_call",
-    "params": [
-      {
-        "to": "0xc582Bc0317dbb0908203541971a358c44b1F3766",
-        "data": "0x0d43e8ad"
-      },
-      "latest"
-    ],
-    "response": {
-      "jsonrpc": "2.0",
-      "id": 5,
-      "result": "0x000000000000000000000000b2b580ce436e6f77a5713d80887e14788ef49c9a"
-    }
   }
 ]
-

pyproject.toml

@@ -25,6 +25,7 @@ more-itertools = "^10.1.0"
 web3 = "^6.10.0"
 web3-multi-provider = "^0.6.0"
 json-stream = "^2.3.2"
+lazy-object-proxy = "^1.9.0"
 
 [tool.poetry.group.dev.dependencies]
 base58 = "^2.1.1"
@@ -60,7 +61,6 @@ branch = true
 [tool.pylint.format]
 max-line-length = "120"
 
-
 [tool.pylint."messages control"]
 disable = [
     # Disabled by default

src/main.py

@@ -20,7 +20,8 @@
     ConsensusClientModule,
     KeysAPIClientModule,
     LidoValidatorsProvider,
-    FallbackProviderModule
+    FallbackProviderModule,
+    LazyCSM
 )
 from src.web3py.middleware import metrics_collector
 from src.web3py.typings import Web3
@@ -40,12 +41,17 @@ def main(module_name: OracleModule):
             'module': module_name,
             'ACCOUNT': variables.ACCOUNT.address if variables.ACCOUNT else 'Dry',
             'LIDO_LOCATOR_ADDRESS': variables.LIDO_LOCATOR_ADDRESS,
+            'CSM_ORACLE_ADDRESS': variables.CSM_ORACLE_ADDRESS,
+            'CSM_MODULE_ADDRESS': variables.CSM_MODULE_ADDRESS,
+            'FINALIZATION_BATCH_MAX_REQUEST_COUNT': variables.FINALIZATION_BATCH_MAX_REQUEST_COUNT,
             'MAX_CYCLE_LIFETIME_IN_SECONDS': variables.MAX_CYCLE_LIFETIME_IN_SECONDS,
         },
     })
     ENV_VARIABLES_INFO.info({
         "ACCOUNT": str(variables.ACCOUNT.address) if variables.ACCOUNT else 'Dry',
         "LIDO_LOCATOR_ADDRESS": str(variables.LIDO_LOCATOR_ADDRESS),
+        "CSM_ORACLE_ADDRESS": str(variables.CSM_ORACLE_ADDRESS),
+        "CSM_MODULE_ADDRESS": str(variables.CSM_MODULE_ADDRESS),
         "FINALIZATION_BATCH_MAX_REQUEST_COUNT": str(variables.FINALIZATION_BATCH_MAX_REQUEST_COUNT),
         "MAX_CYCLE_LIFETIME_IN_SECONDS": str(variables.MAX_CYCLE_LIFETIME_IN_SECONDS),
     })
@@ -78,6 +84,7 @@ def main(module_name: OracleModule):
         'lido_contracts': LidoContracts,
         'lido_validators': LidoValidatorsProvider,
         'transaction': TransactionUtils,
+        'csm': LazyCSM,
         'cc': lambda: cc,  # type: ignore[dict-item]
         'kac': lambda: kac,  # type: ignore[dict-item]
     })

src/modules/csm/csm.py

@@ -1,12 +1,14 @@
 from functools import cached_property
 import logging
 
+from web3.types import BlockIdentifier
+
 from src.metrics.prometheus.business import CONTRACT_ON_PAUSE
 from src.metrics.prometheus.duration_meter import duration_meter
 from src.modules.csm.typings import FramePerformance, ReportData
 from src.modules.submodules.consensus import ConsensusModule
 from src.modules.submodules.oracle_module import BaseModule, ModuleExecuteDelay
-from src.typings import BlockStamp, ReferenceBlockStamp
+from src.typings import BlockStamp, ReferenceBlockStamp, SlotNumber
 from src.utils.cache import global_lru_cache as lru_cache
 from src.web3py.extensions.lido_validators import NodeOperatorId, StakingModule, ValidatorsByNodeOperator
 from src.web3py.typings import Web3
@@ -29,12 +31,12 @@ class CSFeeOracle(BaseModule, ConsensusModule):
     CONTRACT_VERSION = 1
 
     def __init__(self, w3: Web3):
-        self.report_contract = w3.lido_contracts.csm_oracle
+        self.report_contract = w3.csm.oracle
         super().__init__(w3)
         self.frame_performance: FramePerformance | None
 
     def refresh_contracts(self):
-        self.report_contract = self.w3.lido_contracts.csm_oracle
+        self.report_contract = self.w3.csm.oracle
 
     def execute_module(self, last_finalized_blockstamp: BlockStamp) -> ModuleExecuteDelay:
         report_blockstamp = self.get_blockstamp_for_report(last_finalized_blockstamp)
@@ -56,8 +58,8 @@ def build_report(self, blockstamp: ReferenceBlockStamp) -> tuple:
         assert self.frame_performance.is_coherent
 
         # Get the current frame.
-        _ = self.w3.lido_contracts.get_csm_last_processing_ref_slot(blockstamp)
-        _ = self.get_current_frame(blockstamp).ref_slot
+        last_ref_slot = self.w3.csm.get_csm_last_processing_ref_slot(blockstamp)
+        ref_slot = self.get_current_frame(blockstamp).ref_slot
 
         # Get module's node operators.
         _ = self.module_validators_by_node_operators(blockstamp)
@@ -66,17 +68,18 @@ def build_report(self, blockstamp: ReferenceBlockStamp) -> tuple:
         # Build the map of the current distribution operators.
         # _ = groupby(self.frame_performance.aggr_per_val, operators)
         # Exclude validators of operators with stuck keys.
-        events = self.w3.lido_contracts.get_csm_stuck_keys_events(blockstamp)
-        for _ in events:
-            ...
+        _ = self.w3.csm.get_csm_stuck_node_operators(
+            self._slot_to_block_identifier(last_ref_slot),
+            self._slot_to_block_identifier(ref_slot),
+        )
         # Exclude underperforming validators.
 
         # Calculate share of each CSM node operator.
         _ = self._to_distribute(blockstamp)
         shares: tuple[tuple[NodeOperatorId, int]] = tuple()  # type: ignore
 
         # Load the previous tree if any.
-        _ = self.w3.lido_contracts.get_csm_tree_cid(blockstamp)
+        _ = self.w3.csm.get_csm_tree_cid(blockstamp)
         # leafs = []
         # if cid:
         #     leafs = parse_leafs(ipfs.get(cid))
@@ -91,7 +94,7 @@ def build_report(self, blockstamp: ReferenceBlockStamp) -> tuple:
         ).as_tuple()
 
     def is_main_data_submitted(self, blockstamp: BlockStamp) -> bool:
-        last_ref_slot = self.w3.lido_contracts.get_csm_last_processing_ref_slot(blockstamp)
+        last_ref_slot = self.w3.csm.get_csm_last_processing_ref_slot(blockstamp)
         ref_slot = self.get_current_frame(blockstamp).ref_slot
         return last_ref_slot == ref_slot
 
@@ -122,7 +125,7 @@ def _is_paused(self, blockstamp: ReferenceBlockStamp) -> bool:
         return self.report_contract.functions.isPaused().call(block_identifier=blockstamp.block_hash)
 
     def _collect_data(self, blockstamp: BlockStamp) -> None:
-        last_ref_slot = self.w3.lido_contracts.get_csm_last_processing_ref_slot(blockstamp)
+        last_ref_slot = self.w3.csm.get_csm_last_processing_ref_slot(blockstamp)
         ref_slot = self.get_current_frame(blockstamp).ref_slot
 
         # TODO: To think about the proper cache invalidation conditions.
@@ -142,7 +145,12 @@ def _collect_data(self, blockstamp: BlockStamp) -> None:
         self.frame_performance.dump()
 
     def _to_distribute(self, blockstamp: ReferenceBlockStamp) -> int:
-        # TODO: Move away.
-        return self.w3.lido_contracts.csm_distributor.functions.pendingToDistribute().call(
-            block_identifier=blockstamp.block_hash
-        )
+        return self.w3.csm.fee_distributor.pending_to_distribute(blockstamp.block_hash)
+
+    def _slot_to_block_identifier(self, slot: SlotNumber) -> BlockIdentifier:
+        block = self.w3.cc.get_block_details(slot)
+
+        try:
+            return block.message.body["execution_payload"]["block_hash"]
+        except KeyError as e:
+            raise ValueError(f"ExecutionPayload not found in slot {slot}") from e

src/providers/consensus/typings.py

@@ -1,7 +1,8 @@
 from dataclasses import dataclass
 from enum import Enum
+from typing import TypedDict
 
-from src.typings import BlockRoot, StateRoot
+from src.typings import BlockHash, BlockRoot, StateRoot
 from src.utils.dataclass import Nested, FromResponse
 
 
@@ -57,13 +58,24 @@ class BlockHeaderFullResponse(Nested, FromResponse):
     finalized: bool | None = None
 
 
+class ExecutionPayload(TypedDict):
+    parent_hash: BlockHash
+    block_number: int
+    timestamp: int
+    block_hash: BlockHash
+
+
+class BeaconBlockBody(TypedDict, total=False):
+    execution_payload: ExecutionPayload
+
+
 @dataclass
 class BlockMessage(FromResponse):
     slot: str
     proposer_index: str
     parent_root: str
     state_root: StateRoot
-    body: dict
+    body: BeaconBlockBody
 
 
 class ValidatorStatus(Enum):

src/providers/execution/contracts/CSFeeDistributor.py

@@ -0,0 +1,44 @@
+import json
+import logging
+
+from eth_typing.evm import ChecksumAddress
+from web3.contract.contract import Contract
+from web3.types import BlockIdentifier
+
+logger = logging.getLogger(__name__)
+
+
+class CSFeeDistributor(Contract):
+    abi_path = "./assets/CSFeeDistributor.json"
+
+    # TODO: Inherit from the base class.
+    def __init__(self, address: ChecksumAddress | None = None) -> None:
+        with open(self.abi_path, encoding="utf-8") as f:
+            self.abi = json.load(f)
+        super().__init__(address)
+
+    def pending_to_distribute(self, block: BlockIdentifier = "latest") -> int:
+        """Returns the amount of shares that are pending to be distributed"""
+
+        resp = self.functions.pendingToDistribute().call(block_identifier=block)
+        logger.debug(
+            {
+                "msg": "Call to pendingToDistribute()",
+                "value": resp,
+                "block_identifier": repr(block),
+            }
+        )
+        return resp
+
+    def tree_cid(self, block: BlockIdentifier = "latest") -> str:
+        """CID of the latest published Merkle tree"""
+
+        resp = self.functions.treeCid().call(block_identifier=block)
+        logger.debug(
+            {
+                "msg": "Call to treeCid()",
+                "value": resp,
+                "block_identifier": repr(block),
+            }
+        )
+        return resp

src/providers/execution/contracts/CSFeeOracle.py

@@ -0,0 +1,46 @@
+import json
+import logging
+from typing import cast
+
+from eth_typing.evm import Address, ChecksumAddress
+from web3.contract.contract import Contract
+from web3.types import BlockIdentifier
+
+from src.typings import SlotNumber
+
+logger = logging.getLogger(__name__)
+
+
+class CSFeeOracle(Contract):
+    abi_path = "./assets/CSFeeOracle.json"
+
+    # TODO: Inherit from the base class.
+    def __init__(self, address: ChecksumAddress | None = None) -> None:
+        with open(self.abi_path, encoding="utf-8") as f:
+            self.abi = json.load(f)
+        super().__init__(address)
+
+    def fee_distributor(self, block: BlockIdentifier = "latest") -> Address:
+        """Returns the address of the CSFeeDistributor"""
+
+        resp = self.functions.feeDistributor().call(block_identifier=block)
+        logger.debug(
+            {
+                "msg": "Call to feeDistributor()",
+                "value": resp,
+                "block_identifier": repr(block),
+            }
+        )
+        return cast(Address, resp)
+
+    # TODO: Inherit the method from the BaseOracle class.
+    def get_last_processing_ref_slot(self, block: BlockIdentifier = "latest") -> SlotNumber:
+        resp = self.functions.getLastProcessingRefSlot().call(block_identifier=block)
+        logger.debug(
+            {
+                "msg": "Call to getLastProcessingRefSlot()",
+                "value": resp,
+                "block_identifier": repr(block),
+            }
+        )
+        return SlotNumber(resp)