Fix report (#448)

src/modules/csm/checkpoint.py

@@ -1,14 +1,14 @@
 import logging
 import time
-from threading import Lock
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from threading import Lock
 from typing import cast
 
 from timeout_decorator import TimeoutError as DecoratorTimeoutError
 
 from src.modules.csm.state import State
 from src.providers.consensus.client import ConsensusClient
-from src.typings import EpochNumber, BlockRoot, SlotNumber, BlockStamp, ValidatorIndex
+from src.typings import BlockRoot, BlockStamp, EpochNumber, SlotNumber, ValidatorIndex
 from src.utils.range import sequence
 from src.utils.web3converter import Web3Converter
 
@@ -33,20 +33,17 @@ def __init__(self, cc: ConsensusClient, converter: Web3Converter, state: State):
         self.converter = converter
         self.state = state
 
-    def prepare_checkpoints(
-        self,
-        l_epoch: EpochNumber,
-        r_epoch: EpochNumber,
-        finalized_epoch: EpochNumber
-    ):
+    def prepare_checkpoints(self, l_epoch: EpochNumber, r_epoch: EpochNumber, finalized_epoch: EpochNumber):
         def _prepare_checkpoint(_slot: SlotNumber, _duty_epochs: list[EpochNumber]):
             return Checkpoint(self.cc, self.converter, self.state, _slot, _duty_epochs)
 
-        l_epoch = min(self.state.unprocessed_epochs) or l_epoch
-        if l_epoch == r_epoch:
+        if not self.state.unprocessed_epochs:
             logger.info({"msg": "All epochs processed. No checkpoint required."})
             return []
 
+        l_epoch = min(self.state.unprocessed_epochs) or l_epoch
+        assert l_epoch < r_epoch
+
         processing_delay = finalized_epoch - l_epoch
         if processing_delay < self.MIN_CHECKPOINT_STEP and finalized_epoch < r_epoch:
             logger.info(
@@ -92,7 +89,7 @@ def __init__(
         converter: Web3Converter,
         state: State,
         slot: SlotNumber,
-        duty_epochs: list[EpochNumber]
+        duty_epochs: list[EpochNumber],
     ):
         self.cc = cc
         self.converter = converter
@@ -131,15 +128,13 @@ def _unprocessed():
                 ext.shutdown(wait=True, cancel_futures=True)
                 raise ValueError(e) from e
 
-    def _select_roots_to_check(
-        self, duty_epoch: EpochNumber
-    ) -> list[BlockRoot | None]:
+    def _select_roots_to_check(self, duty_epoch: EpochNumber) -> list[BlockRoot | None]:
         # inspired by the spec
         # https://github.com/ethereum/consensus-specs/blob/dev/specs/phase0/beacon-chain.md#get_block_root_at_slot
         roots_to_check = []
         slots = sequence(
             self.converter.get_epoch_first_slot(duty_epoch),
-            self.converter.get_epoch_last_slot(EpochNumber(duty_epoch + 1))
+            self.converter.get_epoch_last_slot(EpochNumber(duty_epoch + 1)),
         )
         for slot_to_check in slots:
             # TODO: get the magic number from the CL spec
@@ -202,9 +197,7 @@ def _process_attestations(self, slot_data: dict, committees: dict) -> None:
         def to_bits(aggregation_bits: str):
             # copied from https://github.com/ethereum/py-ssz/blob/main/ssz/sedes/bitvector.py#L66
             att_bytes = bytes.fromhex(aggregation_bits[2:])
-            return [
-                bool((att_bytes[bit_index // 8] >> bit_index % 8) % 2) for bit_index in range(len(att_bytes) * 8)
-            ]
+            return [bool((att_bytes[bit_index // 8] >> bit_index % 8) % 2) for bit_index in range(len(att_bytes) * 8)]
 
         for attestation in slot_data['message']['body']['attestations']:
             committee_id = f"{attestation['data']['slot']}{attestation['data']['index']}"

src/modules/csm/csm.py

@@ -69,10 +69,11 @@ def execute_module(self, last_finalized_blockstamp: BlockStamp) -> ModuleExecute
     def build_report(self, blockstamp: ReferenceBlockStamp) -> tuple:
         # pylint: disable=too-many-branches,too-many-statements
         assert self.state
-        l_ref_slot, r_ref_slot = self.current_frame_range(blockstamp)
+        l_ref_slot = self.w3.csm.get_csm_last_processing_ref_slot(blockstamp)
+        r_ref_slot = blockstamp.ref_slot
         converter = self.converter(blockstamp)
         l_epoch = EpochNumber(converter.get_epoch_by_slot(l_ref_slot) + 1)
-        r_epoch = converter.get_epoch_by_slot(r_ref_slot)
+        r_epoch = blockstamp.ref_epoch
 
         try:
             self.state.validate_for_report(l_epoch, r_epoch)
@@ -119,7 +120,8 @@ def build_report(self, blockstamp: ReferenceBlockStamp) -> tuple:
                     # is not presented in the aggregates (e.g. exited, pending for activation etc).
                     continue
 
-            distribution[no_id] = portion
+            if portion:
+                distribution[no_id] = portion
 
         # Calculate share of each CSM node operator.
         to_distribute = self.w3.csm.fee_distributor.pending_to_distribute(blockstamp.block_hash)