feat: state data is list of sequences now

Author: vgorkavenko

src/modules/csm/checkpoint.py

@@ -20,8 +20,7 @@
 lock = Lock()
 
 
-class MinStepIsNotReached(Exception):
-    ...
+class MinStepIsNotReached(Exception): ...
 
 
 @dataclass
@@ -33,7 +32,7 @@ class FrameCheckpoint:
 @dataclass
 class ValidatorDuty:
     index: ValidatorIndex
-    included: bool
+    is_included: bool
 
 
 class FrameCheckpointsIterator:
@@ -198,9 +197,10 @@ def _check_duty(
         with lock:
             for committee in committees.values():
                 for validator_duty in committee:
-                    self.state.inc(
+                    self.state.set_duty_status(
+                        duty_epoch,
                         validator_duty.index,
-                        included=validator_duty.included,
+                        validator_duty.is_included,
                     )
             if duty_epoch not in self.state.unprocessed_epochs:
                 raise ValueError(f"Epoch {duty_epoch} is not in epochs that should be processed")
@@ -222,7 +222,7 @@ def _prepare_committees(self, epoch: EpochNumber) -> Committees:
             validators = []
             # Order of insertion is used to track the positions in the committees.
             for validator in committee.validators:
-                validators.append(ValidatorDuty(index=ValidatorIndex(int(validator)), included=False))
+                validators.append(ValidatorDuty(index=ValidatorIndex(int(validator)), is_included=False))
             committees[(committee.slot, committee.index)] = validators
         return committees
 
@@ -233,7 +233,7 @@ def process_attestations(attestations: Iterable[BlockAttestation], committees: C
         committee = committees.get(committee_id, [])
         att_bits = _to_bits(attestation.aggregation_bits)
         for index_in_committee, validator_duty in enumerate(committee):
-            validator_duty.included = validator_duty.included or _is_attested(att_bits, index_in_committee)
+            validator_duty.is_included = validator_duty.is_included or _is_attested(att_bits, index_in_committee)
 
 
 def _is_attested(bits: Sequence[bool], index: int) -> bool:

src/modules/csm/csm.py

@@ -12,7 +12,8 @@
 )
 from src.metrics.prometheus.duration_meter import duration_meter
 from src.modules.csm.checkpoint import FrameCheckpointProcessor, FrameCheckpointsIterator, MinStepIsNotReached
-from src.modules.csm.log import FramePerfLog
+from src.modules.csm.duties.attestation import calc_performance
+from src.modules.csm.log import FramePerfLog, AttestationsAccumulatorLog
 from src.modules.csm.state import State
 from src.modules.csm.tree import Tree
 from src.modules.csm.types import ReportData, Shares
@@ -33,7 +34,7 @@
 )
 from src.utils.blockstamp import build_blockstamp
 from src.utils.cache import global_lru_cache as lru_cache
-from src.utils.slot import get_next_non_missed_slot
+from src.utils.slot import get_next_non_missed_slot, get_reference_blockstamp
 from src.utils.web3converter import Web3Converter
 from src.web3py.extensions.lido_validators import NodeOperatorId, StakingModule, ValidatorsByNodeOperator
 from src.web3py.types import Web3
@@ -102,12 +103,12 @@ def build_report(self, blockstamp: ReferenceBlockStamp) -> tuple:
         if (prev_cid is None) != (prev_root == ZERO_HASH):
             raise InconsistentData(f"Got inconsistent previous tree data: {prev_root=} {prev_cid=}")
 
-        distributed, shares, log = self.calculate_distribution(blockstamp)
+        distributed, shares, logs = self.calculate_distribution(blockstamp)
 
         if distributed != sum(shares.values()):
             raise InconsistentData(f"Invalid distribution: {sum(shares.values())=} != {distributed=}")
 
-        log_cid = self.publish_log(log)
+        log_cid = self.publish_log(logs)
 
         if not distributed and not shares:
             logger.info({"msg": "No shares distributed in the current frame"})
@@ -225,27 +226,64 @@ def collect_data(self, blockstamp: BlockStamp) -> bool:
 
     def calculate_distribution(
         self, blockstamp: ReferenceBlockStamp
-    ) -> tuple[int, defaultdict[NodeOperatorId, int], FramePerfLog]:
+    ) -> tuple[int, defaultdict[NodeOperatorId, int], list[FramePerfLog]]:
         """Computes distribution of fee shares at the given timestamp"""
-
-        network_avg_perf = self.state.get_network_aggr().perf
-        threshold = network_avg_perf - self.w3.csm.oracle.perf_leeway_bp(blockstamp.block_hash) / TOTAL_BASIS_POINTS
         operators_to_validators = self.module_validators_by_node_operators(blockstamp)
 
+        distributed = 0
+        # Calculate share of each CSM node operator.
+        shares = defaultdict[NodeOperatorId, int](int)
+        logs: list[FramePerfLog] = []
+
+        converter = self.converter(blockstamp)
+        frames = self.state.calc_frames(converter.frame_config.epochs_per_frame)
+        for from_epoch, to_epoch in frames:
+            frame_blockstamp = blockstamp
+            frame_ref_slot = converter.get_epoch_first_slot(to_epoch)
+            if blockstamp.slot_number != frame_ref_slot:
+                frame_blockstamp = get_reference_blockstamp(
+                    cc=self.w3.cc,
+                    ref_slot=converter.get_epoch_first_slot(to_epoch),
+                    ref_epoch=to_epoch,
+                    last_finalized_slot_number=blockstamp.slot_number,
+                )
+            distributed_in_frame, shares_in_frame, log = self._calculate_distribution_in_frame(
+                frame_blockstamp, operators_to_validators, from_epoch, to_epoch, distributed
+            )
+            distributed += distributed_in_frame
+            for no_id, share in shares_in_frame.items():
+                shares[no_id] += share
+            logs.append(log)
+
+        return distributed, shares, logs
+
+    def _calculate_distribution_in_frame(
+        self,
+        blockstamp: ReferenceBlockStamp,
+        operators_to_validators: ValidatorsByNodeOperator,
+        from_epoch: EpochNumber,
+        to_epoch: EpochNumber,
+        distributed: int,
+    ):
+        network_perf = self.state.calc_network_perf(from_epoch, to_epoch)
+        threshold = network_perf - self.w3.csm.oracle.perf_leeway_bp(blockstamp.block_hash) / TOTAL_BASIS_POINTS
+
         # Build the map of the current distribution operators.
         distribution: dict[NodeOperatorId, int] = defaultdict(int)
         stuck_operators = self.stuck_operators(blockstamp)
-        log = FramePerfLog(blockstamp, self.state.frame, threshold)
+        log = FramePerfLog(blockstamp, (from_epoch, to_epoch), threshold)
 
         for (_, no_id), validators in operators_to_validators.items():
             if no_id in stuck_operators:
                 log.operators[no_id].stuck = True
                 continue
 
             for v in validators:
-                aggr = self.state.data.get(ValidatorIndex(int(v.index)))
+                missed = self.state.count_missed(ValidatorIndex(int(v.index)), from_epoch, to_epoch)
+                included = self.state.count_included(ValidatorIndex(int(v.index)), from_epoch, to_epoch)
+                assigned = missed + included
 
-                if aggr is None:
+                if not assigned:
                     # It's possible that the validator is not assigned to any duty, hence it's performance
                     # is not presented in the aggregates (e.g. exited, pending for activation etc).
                     continue
@@ -256,23 +294,23 @@ def calculate_distribution(
                     log.operators[no_id].validators[v.index].slashed = True
                     continue
 
-                if aggr.perf > threshold:
+                perf = calc_performance(included, missed)
+                if perf > threshold:
                     # Count of assigned attestations used as a metrics of time
                     # the validator was active in the current frame.
-                    distribution[no_id] += aggr.assigned
+                    distribution[no_id] += assigned
 
-                log.operators[no_id].validators[v.index].perf = aggr
+                log.operators[no_id].validators[v.index].perf = AttestationsAccumulatorLog(assigned, included)
 
         # Calculate share of each CSM node operator.
         shares = defaultdict[NodeOperatorId, int](int)
         total = sum(p for p in distribution.values())
+        to_distribute = self.w3.csm.fee_distributor.shares_to_distribute(blockstamp.block_hash) - distributed
+        log.distributable = to_distribute
 
         if not total:
             return 0, shares, log
 
-        to_distribute = self.w3.csm.fee_distributor.shares_to_distribute(blockstamp.block_hash)
-        log.distributable = to_distribute
-
         for no_id, no_share in distribution.items():
             if no_share:
                 shares[no_id] = to_distribute * no_share // total
@@ -343,9 +381,9 @@ 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, log: FramePerfLog) -> CID:
-        log_cid = self.w3.ipfs.publish(log.encode())
-        logger.info({"msg": "Frame log uploaded to IPFS", "cid": repr(log_cid)})
+    def publish_log(self, logs: list[FramePerfLog]) -> CID:
+        log_cid = self.w3.ipfs.publish(FramePerfLog.encode(logs))
+        logger.info({"msg": "Frame(s) log uploaded to IPFS", "cid": repr(log_cid)})
         return log_cid
 
     @lru_cache(maxsize=1)

src/modules/csm/duties/attestation.py

@@ -0,0 +1,60 @@
+from enum import Enum
+from typing import NewType
+
+EpochIndexInFrame = NewType('EpochIndexInFrame', int)
+
+
+class AttestationStatus(Enum):
+    NO_DUTY = None
+    MISSED = False
+    INCLUDED = True
+
+
+def calc_performance(included: int, missed: int) -> float:
+    all_ = missed + included
+    return included / all_ if all_ else 0
+
+
+class AttestationSequence(list):
+    """
+    Duties sequence for a CSM performance oracle frame for validators.
+
+    It is bits sequence where each pair of bits represents the duty status of a validator for a specific epoch:
+     - None - no duty
+     - False - missed. assigned but not included attestation
+     - True - included attestation
+
+    Every index in the sequence corresponds to epoch index in report frame.
+    For example:
+     Report frame is [100000 epoch, ..., 100510 epoch],
+     We need to write duty status for epoch 100000 at index 0, for epoch 100001 at index 1, and so on.
+    """
+
+    def __init__(self, size: int):
+        super().__init__([AttestationStatus.NO_DUTY] * size)
+
+    def __str__(self):
+        missed, included = self.count_missed(), self.count_included()
+        return f"{self.__class__.__name__}({missed=}, {included=})"
+
+    def _validate_range(self, from_index: EpochIndexInFrame, to_index: EpochIndexInFrame):
+        if from_index < 0 or to_index > len(self) or from_index >= to_index:
+            raise ValueError("Invalid range for from_index and to_index")
+
+    def count_missed(self, from_index: EpochIndexInFrame = 0, to_index: EpochIndexInFrame = None):
+        if to_index is None:
+            to_index = len(self)
+        self._validate_range(from_index, to_index)
+        return self[from_index:to_index].count(AttestationStatus.MISSED)
+
+    def count_included(self, from_index: EpochIndexInFrame = 0, to_index: EpochIndexInFrame = None):
+        if to_index is None:
+            to_index = len(self)
+        self._validate_range(from_index, to_index)
+        return self[from_index:to_index].count(AttestationStatus.INCLUDED)
+
+    def get_duty_status(self, epoch_index: EpochIndexInFrame) -> AttestationStatus:
+        return self[epoch_index]
+
+    def set_duty_status(self, epoch_index: EpochIndexInFrame, duty_status: AttestationStatus):
+        self[epoch_index] = duty_status

src/modules/csm/log.py

@@ -2,17 +2,23 @@
 from collections import defaultdict
 from dataclasses import asdict, dataclass, field
 
-from src.modules.csm.state import AttestationsAccumulator
 from src.modules.csm.types import Shares
 from src.types import EpochNumber, NodeOperatorId, ReferenceBlockStamp
 
 
 class LogJSONEncoder(json.JSONEncoder): ...
 
 
+@dataclass
+class AttestationsAccumulatorLog:
+    assigned: int = 0
+    included: int = 0
+
+
 @dataclass
 class ValidatorFrameSummary:
-    perf: AttestationsAccumulator = field(default_factory=AttestationsAccumulator)
+    # TODO: Should be renamed. Perf means different things in different contexts
+    perf: AttestationsAccumulatorLog = field(default_factory=AttestationsAccumulatorLog)
     slashed: bool = False
 
 
@@ -35,13 +41,14 @@ class FramePerfLog:
         default_factory=lambda: defaultdict(OperatorFrameSummary)
     )
 
-    def encode(self) -> bytes:
+    @staticmethod
+    def encode(logs: list['FramePerfLog']) -> bytes:
         return (
             LogJSONEncoder(
                 indent=None,
                 separators=(',', ':'),
                 sort_keys=True,
             )
-            .encode(asdict(self))
+            .encode([asdict(log) for log in logs])
             .encode()
         )