fix: fast updates sampling, crashing, registration spam

Author: janezicmatej

observer/fast_updates_manager.py

@@ -1,4 +1,3 @@
-from collections import deque
 from collections.abc import Sequence
 
 from attrs import define, field, frozen
@@ -20,7 +19,8 @@ class FastUpdate:
 
 @define
 class FastUpdatesManager:
-    fast_updates: deque[FastUpdate] = field(factory=deque)
+    last_update_block: int
+    last_update: FastUpdate
     address_list: set[ChecksumAddress] = field(factory=set)
 
     async def check_addresses(
@@ -36,30 +36,24 @@ def check_update_length(
         mb = Message.builder()
         messages = []
         level = MessageLevel.WARNING
-        fus = list(
-            filter(
-                lambda x: x.reward_epoch_id >= fast_update_re
-                and x.address in self.address_list,
-                self.fast_updates,
-            )
-        )
-        if len(fus) > 0:
-            fu = fus[-1]
-        else:
+
+        if self.last_update.reward_epoch_id != fast_update_re or not nr_of_feeds:
             return messages
-        rounded_nr_feeds = nr_of_feeds
-        # update arrays are whole bytes, so we need to round the number of feeds
-        # up to the nearest multiple of 8 if it is not one already
-        if nr_of_feeds % 8 != 0:
-            rounded_nr_feeds = nr_of_feeds + (8 - nr_of_feeds % 8)
-        if rounded_nr_feeds > 0 and len(fu.update_array) != rounded_nr_feeds:
+
+        # round up to the next multiple of 8 as update array is encoded in integer
+        # number of bytes
+        expected = nr_of_feeds + 7 - (nr_of_feeds - 1) % 8
+        submitted = len(self.last_update.update_array)
+
+        if expected != submitted:
             messages.append(
                 mb.build(
                     level,
                     (
-                        "Incorrect length of last update array, should be"
-                        f" {rounded_nr_feeds} but got {len(fu.update_array)}"
+                        f"incorrect number of sent feeds, should be {expected} but got "
+                        f"{submitted}"
                     ),
                 )
             )
+
         return messages

observer/observer.py

@@ -396,7 +396,10 @@ async def observer_loop(config: Configuration) -> None:
     contracts = cm.get_contracts_list()
     event_signatures = cm.get_events()
 
-    fum = FastUpdatesManager()
+    entity = signing_policy.entity_mapper.by_identity_address[tia]
+    fum = FastUpdatesManager(
+        block_number, FastUpdate(reward_epoch.id, entity.signing_policy_address, [])
+    )
     spm = SigningPolicyManager(signing_policy, signing_policy)
     rm = RewardManager()
 
@@ -418,6 +421,7 @@ async def observer_loop(config: Configuration) -> None:
     )
     last_minimal_conditions_check = int(time.time())
     last_ping = time.time()
+    last_registration_check = time.time()
 
     node_connections = defaultdict(deque)
     uptime_validations = 0
@@ -431,10 +435,6 @@ async def observer_loop(config: Configuration) -> None:
     voter_registration_started_ts: int = 0
     registered: bool = False
 
-    preregistration_started: bool = False
-    preregistration_started_ts: int = 0
-    preregistered: bool = False
-
     nr_of_feeds: int = 0
     fast_update_re: int = 0
 
@@ -528,6 +528,8 @@ async def observer_loop(config: Configuration) -> None:
                         case "VoterRegistered":
                             e = VoterRegistered.from_dict(data["args"])
                             spb.add(e)
+                            if registered:
+                                continue
                             entity = signing_policy.entity_mapper.by_identity_address[
                                 tia
                             ]
@@ -543,73 +545,51 @@ async def observer_loop(config: Configuration) -> None:
                             e = VoterRegistrationInfo.from_dict(data["args"])
                             spb.add(e)
                         case "VotePowerBlockSelected":
-                            preregistration_started = False
-                            preregistered = False
                             e = VotePowerBlockSelected.from_dict(data["args"])
                             spb.add(e)
+                            if registered:
+                                continue
                             voter_registration_started = True
                             voter_registration_started_ts = int(time.time())
                         case "RandomAcquisitionStarted":
                             e = RandomAcquisitionStarted.from_dict(data["args"])
                             spb.add(e)
-                            preregistration_started = True
-                            preregistration_started_ts = int(time.time())
                         case "FastUpdateFeedsSubmitted":
-                            e = FastUpdateFeedsSubmitted.from_dict(
-                                data["args"],
-                                data["address"],
-                                data["transactionHash"],
-                            )
+                            e = FastUpdateFeedsSubmitted.from_dict(data)
                             tx = await w.eth.get_transaction(e.transaction_hash)
                             spa, address, update_array = calculate_update_from_tx(
                                 config, w, tx
                             )
                             entity = signing_policy.entity_mapper.by_identity_address[
                                 tia
                             ]
-                            fum.fast_updates.append(
-                                FastUpdate(
+                            if un_prefix_0x(entity.signing_policy_address) == spa:
+                                fum.last_update = FastUpdate(
                                     signing_policy.reward_epoch.id,
                                     address,
                                     update_array,
                                 )
-                            )
-                            # with expected sample size 1 and average block time of
-                            # 1s, this should be an ok approximation
-                            if (
-                                len(fum.fast_updates)
-                                > minimal_conditions.time_period.value
-                            ):
-                                fum.fast_updates.popleft()
-                            if un_prefix_0x(entity.signing_policy_address) == spa:
+                                fum.last_update_block = int(data["blockNumber"])
                                 # We check update array when we receive a new one
                                 event_messages.extend(
                                     fum.check_update_length(nr_of_feeds, fast_update_re)
                                 )
                                 fum.address_list.add(address)
                         case "FastUpdateFeeds":
-                            e = FastUpdateFeeds.from_dict(
-                                data["args"],
-                                data["address"],
-                                data["transactionHash"],
-                            )
+                            e = FastUpdateFeeds.from_dict(data)
                             nr_of_feeds, fast_update_re = (
                                 len(e.feeds),
                                 ref.from_voting_epoch(
                                     vef.make_epoch(e.voting_round_id)
                                 ).id,
                             )
                         case "VoterPreRegistered":
-                            e = VoterPreRegistered.from_dict(
-                                data["args"],
-                                data["address"],
-                                data["transactionHash"],
-                            )
+                            e = VoterPreRegistered.from_dict(data)
                             entity = signing_policy.entity_mapper.by_identity_address[
                                 tia
                             ]
                             if tia == e.voter:
-                                preregistered = True
+                                registered = True
 
             for tx in block_data["transactions"]:
                 assert not isinstance(tx, bytes)
@@ -692,7 +672,7 @@ async def observer_loop(config: Configuration) -> None:
 
                 min_cond_messages.extend(
                     minimal_conditions.calculate_ftso_block_latency_feeds(
-                        entity, spm, fum
+                        entity, signing_policy, fum.last_update_block, block
                     )
                 )
                 min_cond_messages.extend(
@@ -784,23 +764,16 @@ async def observer_loop(config: Configuration) -> None:
             while len(signatures) > minimal_conditions.time_period.value // 90:
                 signatures.popleft()
 
-            # reporting on registration and preregistration
-            if preregistration_started and not preregistered:
-                mb = Message.builder()
-                if int(time.time() - preregistration_started_ts) > 60:
-                    level = MessageLevel.CRITICAL
-                    message = mb.build(
-                        level,
-                        (
-                            "Voter not preregistered after "
-                            f"{int(time.time() - preregistration_started_ts) // 60}"
-                            " minutes"
-                        ),
-                    )
-                    messages.append(message)
-
-            if voter_registration_started and not registered:
-                mb = Message.builder()
+            # reporting on registration and preregistration once per minute
+            interval = 60
+            if int(time.time() - voter_registration_started_ts) > 15 * 60:
+                interval = 10
+            if (
+                int(time.time() - last_registration_check) > interval
+                and voter_registration_started
+                and not registered
+            ):
+                mb = Message.builder().add(network=config.chain_id)
                 if int(time.time() - voter_registration_started_ts) > 60:
                     level = MessageLevel.CRITICAL
                     message = mb.build(

observer/types.py

@@ -191,7 +191,11 @@ class FastUpdateFeedsSubmitted:
     signing_policy_address: ChecksumAddress
 
     @classmethod
-    def from_dict(cls, d: dict[str, Any], address: ChecksumAddress, tx_hash: HexBytes):
+    def from_dict(cls, data: EventData):
+        d = data["args"]
+        address = data["address"]
+        tx_hash = data["transactionHash"]
+
         return cls(
             voting_round_id=int(d["votingRoundId"]),
             emitter_address=address,
@@ -209,7 +213,11 @@ class FastUpdateFeeds:
     decimals: list[int]
 
     @classmethod
-    def from_dict(cls, d: dict[str, Any], address: ChecksumAddress, tx_hash: HexBytes):
+    def from_dict(cls, data: EventData):
+        d = data["args"]
+        address = data["address"]
+        tx_hash = data["transactionHash"]
+
         return cls(
             voting_round_id=int(d["votingEpochId"]),
             emitter_address=address,
@@ -228,9 +236,14 @@ class VoterPreRegistered:
     reward_epoch_id: int
 
     @classmethod
-    def from_dict(cls, d: dict[str, Any], address: ChecksumAddress, tx_hash: HexBytes):
+    def from_dict(cls, data: EventData):
+        d = data["args"]
+        block = data["blockNumber"]
+        address = data["address"]
+        tx_hash = data["transactionHash"]
+
         return cls(
-            block=int(d["block"]),
+            block=int(block),
             emitter_address=address,
             transaction_hash=tx_hash,
             voter=d["voter"],