[WIP] CM v2 - CSM v3 Support

src/constants.py

@@ -62,5 +62,6 @@
 
 ALLOWED_KAPI_VERSION = Version('1.5.0')
 CSM_STATE_VERSION = 1
+CSM_LOGS_VERSION = 1
 
 GENESIS_VALIDATORS_ROOT = bytes([0] * 32)  # all zeros for deposits

src/modules/csm/csm.py

@@ -20,7 +20,7 @@
     StrikesValidator,
 )
 from src.modules.csm.helpers.last_report import LastReport
-from src.modules.csm.log import FramePerfLog
+from src.modules.csm.log import Logs
 from src.modules.csm.state import State
 from src.modules.csm.tree import RewardsTree, StrikesTree, Tree
 from src.modules.csm.types import ReportData, RewardsShares, StrikesList
@@ -136,6 +136,7 @@ def _get_last_report(self, blockstamp: ReferenceBlockStamp) -> LastReport:
         current_frame = self.get_frame_number_by_slot(blockstamp)
         return LastReport.load(self.w3, blockstamp, current_frame)
 
+    @lru_cache(maxsize=1)
     def calculate_distribution(self, blockstamp: ReferenceBlockStamp, last_report: LastReport) -> DistributionResult:
         distribution = Distribution(self.w3, self.converter(blockstamp), self.state)
         result = distribution.calculate(blockstamp, last_report)
@@ -253,8 +254,8 @@ def publish_tree(self, tree: Tree) -> CID:
         logger.info({"msg": "Tree dump uploaded to IPFS", "cid": repr(tree_cid)})
         return tree_cid
 
-    def publish_log(self, logs: list[FramePerfLog]) -> CID:
-        log_cid = self.w3.ipfs.publish(FramePerfLog.encode(logs))
+    def publish_log(self, logs: Logs) -> CID:
+        log_cid = self.w3.ipfs.publish(logs.encode())
         logger.info({"msg": "Frame(s) log uploaded to IPFS", "cid": repr(log_cid)})
         return log_cid
 

src/modules/csm/distribution.py

@@ -4,8 +4,9 @@
 from copy import deepcopy
 from dataclasses import dataclass, field
 
+from src.constants import MIN_ACTIVATION_BALANCE, MAX_EFFECTIVE_BALANCE_ELECTRA, EFFECTIVE_BALANCE_INCREMENT
 from src.modules.csm.helpers.last_report import LastReport
-from src.modules.csm.log import FramePerfLog, OperatorFrameSummary
+from src.modules.csm.log import FramePerfLog, OperatorFrameSummary, Logs
 from src.modules.csm.state import Frame, State, ValidatorDuties
 from src.modules.csm.types import (
     ParticipationShares,
@@ -48,7 +49,7 @@ class DistributionResult:
     total_rebate: RewardsShares = 0
     total_rewards_map: dict[NodeOperatorId, RewardsShares] = field(default_factory=lambda: defaultdict(RewardsShares))
     strikes: dict[StrikesValidator, StrikesList] = field(default_factory=lambda: defaultdict(StrikesList))
-    logs: list[FramePerfLog] = field(default_factory=list)
+    logs: Logs = field(default_factory=Logs)
 
 
 class Distribution:
@@ -99,7 +100,7 @@ def calculate(self, blockstamp: ReferenceBlockStamp, last_report: LastReport) ->
             for no_id, rewards in rewards_map_in_frame.items():
                 result.total_rewards_map[no_id] += rewards
 
-            result.logs.append(frame_log)
+            result.logs.frames.append(frame_log)
 
         if result.total_rewards != sum(result.total_rewards_map.values()):
             raise InconsistentData(
@@ -157,7 +158,10 @@ def _calculate_distribution_in_frame(
             curve_params = self.w3.csm.get_curve_params(no_id, blockstamp)
             log_operator.performance_coefficients = curve_params.perf_coeffs
 
-            active_validators.sort(key=lambda v: v.index)
+            # Sort from biggest to smallest balance and by index from oldest to newest.
+            active_validators.sort(
+                key=lambda v: (-min(v.validator.effective_balance, MAX_EFFECTIVE_BALANCE_ELECTRA), v.index)
+            )
             numbered_validators = enumerate(active_validators, 1)
             for key_number, validator in numbered_validators:
                 key_threshold = max(network_perf - curve_params.perf_leeway_data.get_for(key_number), 0)
@@ -254,10 +258,22 @@ def get_validator_duties_outcome(
             #    87.55 ≈ 88 of 103 participation shares should be counted for the operator key's reward.
             #    The rest 15 participation shares should be counted for the protocol's rebate.
             #
-            participation_share = math.ceil(duties.attestation.assigned * reward_share)
-            rebate_share = duties.attestation.assigned - participation_share
+            val_effective_balance = min(
+                max(MIN_ACTIVATION_BALANCE, validator.validator.effective_balance),
+                MAX_EFFECTIVE_BALANCE_ELECTRA
+            )
+            participation_share_multiplier = val_effective_balance // EFFECTIVE_BALANCE_INCREMENT
+            #
+            # Due to CL rewarding process, validators are getting rewards in proportion to their effective balance:
+            # https://github.com/ethereum/consensus-specs/blob/master/specs/phase0/beacon-chain.md#helpers
+            # Distribution should calculate participation and rebate shares proportional to effective balance as well.
+            # NOTE: Moment of getting 2048 ETH can be too close to the report and trick the distribution.
+            #
+            assigned_att = duties.attestation.assigned * participation_share_multiplier
+            participation_share = math.ceil(assigned_att * reward_share)
+            rebate_share = assigned_att - participation_share
             if rebate_share < 0:
-                raise ValueError(f"Invalid rebate share: {rebate_share=}")
+                raise ValueError(f"Invalid rebate share for validator {validator.index}: {rebate_share=}")
             return ValidatorDutiesOutcome(participation_share, rebate_share, strikes=0)
 
         # In case of bad performance the validator should be striked and assigned attestations are not counted for

src/modules/csm/log.py

@@ -2,6 +2,7 @@
 from collections import defaultdict
 from dataclasses import asdict, dataclass, field
 
+from src.constants import CSM_LOGS_VERSION
 from src.modules.csm.state import DutyAccumulator
 from src.modules.csm.types import RewardsShares
 from src.providers.execution.contracts.cs_parameters_registry import PerformanceCoefficients
@@ -43,14 +44,19 @@ class FramePerfLog:
         default_factory=lambda: defaultdict(OperatorFrameSummary)
     )
 
-    @staticmethod
-    def encode(logs: list['FramePerfLog']) -> bytes:
+
+@dataclass
+class Logs:
+    frames: list[FramePerfLog] = field(default_factory=list)
+    _ver: int = CSM_LOGS_VERSION
+
+    def encode(self) -> bytes:
         return (
             LogJSONEncoder(
                 indent=None,
                 separators=(',', ':'),
                 sort_keys=True,
             )
-            .encode([asdict(log) for log in logs])
+            .encode(asdict(self))
             .encode()
         )

src/providers/execution/contracts/cs_accounting.py

@@ -18,10 +18,10 @@ class CSAccountingContract(ContractInterface):
     def fee_distributor(self, block_identifier: BlockIdentifier = "latest") -> ChecksumAddress:
         """Returns the address of the CSFeeDistributor contract"""
 
-        resp = self.functions.feeDistributor().call(block_identifier=block_identifier)
+        resp = self.functions.FEE_DISTRIBUTOR().call(block_identifier=block_identifier)
         logger.info(
             {
-                "msg": "Call `feeDistributor()`.",
+                "msg": "Call `FEE_DISTRIBUTOR()`.",
                 "value": resp,
                 "block_identifier": repr(block_identifier),
             }