Dynamic blocks (#887)

* Separate events retrieval and their handling.
* Adjust handling of retry queue flag.
* Fix ExchangeCreated handling.

Author: calina-c

aquarius/events/events_monitor.py

@@ -6,6 +6,7 @@
 import logging
 import os
 import time
+from distutils.util import strtobool
 from threading import Thread
 
 import elasticsearch
@@ -147,7 +148,9 @@ def do_run_monitor(self):
 
     def process_current_blocks(self):
         """Process all blocks from the last processed block to the current block."""
-        if os.getenv("PROCESS_RETRY_QUEUE"):
+        process_queue = strtobool(os.getenv("PROCESS_RETRY_QUEUE", "0"))
+
+        if process_queue:
             self.retry_mechanism.process_queue()
 
         last_block = self.get_last_processed_block()
@@ -188,28 +191,73 @@ def process_block_range(self, from_block, to_block):
             self._chain_id,
         ]
 
-        event_processors = {
-            "EVENT_METADATA_CREATED": MetadataCreatedProcessor,
-            "EVENT_METADATA_UPDATED": MetadataUpdatedProcessor,
-            "EVENT_METADATA_STATE": MetadataStateProcessor,
+        # event retrieval
+        all_events = self.get_all_events(from_block, to_block)
+
+        regular_event_processors = {
+            EventTypes.EVENT_METADATA_CREATED: MetadataCreatedProcessor,
+            EventTypes.EVENT_METADATA_UPDATED: MetadataUpdatedProcessor,
+            EventTypes.EVENT_METADATA_STATE: MetadataStateProcessor,
         }
-        for event_name in event_processors:
-            self.handle_regular_event_processor(
-                event_name,
-                event_processors[event_name],
-                processor_args,
-                from_block,
-                to_block,
-            )
 
-        self.handle_price_change(from_block, to_block)
-        self.handle_token_uri_update(from_block, to_block)
-        self.handle_transfer_ownership(from_block, to_block)
+        # event handling
+        for event_name, events_to_process in all_events.items():
+            if event_name == EventTypes.EVENT_TRANSFER:
+                self.handle_transfer_ownership(events_to_process)
+            elif event_name in regular_event_processors.keys():
+                self.handle_regular_event_processor(
+                    event_name,
+                    regular_event_processors[event_name],
+                    processor_args,
+                    events_to_process,
+                )
+            elif event_name in [
+                EventTypes.EVENT_ORDER_STARTED,
+                EventTypes.EVENT_EXCHANGE_CREATED,
+                EventTypes.EVENT_EXCHANGE_RATE_CHANGED,
+                EventTypes.EVENT_DISPENSER_CREATED,
+            ]:
+                self.handle_price_change(event_name, events_to_process, to_block)
+            elif event_name == EventTypes.EVENT_TOKEN_URI_UPDATE:
+                self.handle_token_uri_update(events_to_process)
 
         self.store_last_processed_block(to_block)
 
+    def get_all_events(self, from_block, to_block):
+        all_events = {
+            EventTypes.EVENT_TRANSFER: [],
+            # "regular" events
+            EventTypes.EVENT_METADATA_CREATED: [],
+            EventTypes.EVENT_METADATA_UPDATED: [],
+            EventTypes.EVENT_METADATA_STATE: [],
+            # price changed events
+            EventTypes.EVENT_ORDER_STARTED: [],
+            EventTypes.EVENT_EXCHANGE_CREATED: [],
+            EventTypes.EVENT_EXCHANGE_RATE_CHANGED: [],
+            EventTypes.EVENT_DISPENSER_CREATED: [],
+            #
+            EventTypes.EVENT_TOKEN_URI_UPDATE: [],
+        }
+
+        if from_block >= to_block:
+            return all_events
+
+        try:
+            for event_name in all_events.keys():
+                all_events[event_name] = self.get_event_logs(
+                    event_name, from_block, to_block
+                )
+
+            return all_events
+        except Exception:
+            middle = int((from_block + to_block) / 2)
+            return merge_list_dictionary(
+                self.get_all_events(from_block, middle),
+                self.get_all_events(middle + 1, to_block),
+            )
+
     def handle_regular_event_processor(
-        self, event_name, processor, processor_args, from_block, to_block
+        self, event_name, processor, processor_args, events
     ):
         """Process emitted events between two given blocks for a given event name.
 
@@ -220,19 +268,17 @@ def handle_regular_event_processor(
             from_block (int): inital block
             to_block (int): final block
         """
-        for event in self.get_event_logs(
-            EventTypes.get_value(event_name), from_block, to_block
-        ):
+        for event in events:
             dt_contract = self._web3.eth.contract(
                 abi=ERC721Template.abi,
                 address=self._web3.toChecksumAddress(event.address),
             )
             receipt = self._web3.eth.get_transaction_receipt(
                 event.transactionHash.hex()
             )
-            event_object = dt_contract.events[
-                EventTypes.get_value(event_name)
-            ]().processReceipt(receipt, errors=DISCARD)[0]
+            event_object = dt_contract.events[event_name]().processReceipt(
+                receipt, errors=DISCARD
+            )[0]
             try:
                 metadata_proofs = dt_contract.events.MetadataValidated().processReceipt(
                     receipt, errors=DISCARD
@@ -247,77 +293,67 @@ def handle_regular_event_processor(
                     event.transactionHash.hex(), 0, processor_args[4]
                 )
                 logger.exception(
-                    f"Error processing {EventTypes.get_value(event_name)} event: {e}\n"
-                    f"event={event}"
+                    f"Error processing {event_name} event: {e}\n" f"event={event}"
                 )
 
-    def handle_price_change(self, from_block, to_block):
+    def handle_price_change(self, event_name, events, to_block):
         fre = get_fre(self._web3, self._chain_id)
         dispenser = get_dispenser(self._web3, self._chain_id)
 
-        for event_name in [
-            EventTypes.EVENT_ORDER_STARTED,
-            EventTypes.EVENT_EXCHANGE_CREATED,
-            EventTypes.EVENT_EXCHANGE_RATE_CHANGED,
-            EventTypes.EVENT_DISPENSER_CREATED,
-        ]:
-            events = self.get_event_logs(event_name, from_block, to_block)
-
-            for event in events:
-                if event_name == EventTypes.EVENT_EXCHANGE_CREATED:
-                    receipt = self._web3.eth.get_transaction_receipt(
-                        event.transactionHash.hex()
-                    )
-                    erc20_address = receipt.to
-                elif event_name == EventTypes.EVENT_EXCHANGE_RATE_CHANGED:
-                    receipt = self._web3.eth.get_transaction_receipt(
-                        event.transactionHash.hex()
-                    )
-                    exchange_id = (
-                        fre.events.ExchangeRateChanged()
-                        .processReceipt(receipt)[0]
-                        .args.exchangeId
-                    )
-                    erc20_address = fre.caller.getExchange(exchange_id)[1]
-                elif event_name == EventTypes.EVENT_DISPENSER_CREATED:
-                    receipt = self._web3.eth.get_transaction_receipt(
-                        event.transactionHash.hex()
-                    )
-                    erc20_address = (
-                        dispenser.events.DispenserCreated()
-                        .processReceipt(receipt)[0]
-                        .args.datatokenAddress
-                    )
-                else:
-                    erc20_address = event.address
-
-                erc20_contract = self._web3.eth.contract(
-                    abi=ERC20Template.abi,
-                    address=self._web3.toChecksumAddress(erc20_address),
+        for event in events:
+            if event_name == EventTypes.EVENT_EXCHANGE_CREATED:
+                receipt = self._web3.eth.get_transaction_receipt(
+                    event.transactionHash.hex()
                 )
-
-                logger.debug(
-                    f"{event_name} detected on ERC20 contract {event.address}."
+                exchange_id = (
+                    fre.events.ExchangeCreated()
+                    .processReceipt(receipt)[0]
+                    .args.exchangeId
+                )
+                erc20_address = fre.caller.getExchange(exchange_id)[1]
+            elif event_name == EventTypes.EVENT_EXCHANGE_RATE_CHANGED:
+                receipt = self._web3.eth.get_transaction_receipt(
+                    event.transactionHash.hex()
                 )
+                exchange_id = (
+                    fre.events.ExchangeRateChanged()
+                    .processReceipt(receipt)[0]
+                    .args.exchangeId
+                )
+                erc20_address = fre.caller.getExchange(exchange_id)[1]
+            elif event_name == EventTypes.EVENT_DISPENSER_CREATED:
+                receipt = self._web3.eth.get_transaction_receipt(
+                    event.transactionHash.hex()
+                )
+                erc20_address = (
+                    dispenser.events.DispenserCreated()
+                    .processReceipt(receipt)[0]
+                    .args.datatokenAddress
+                )
+            else:
+                erc20_address = event.address
 
-                try:
-                    event_processor = OrderStartedProcessor(
-                        erc20_contract.caller.getERC721Address(),
-                        self._es_instance,
-                        to_block,
-                        self._chain_id,
-                    )
-                    event_processor.process()
-                except Exception as e:
-                    logger.error(
-                        f"Error processing {event_name} event: {e}\n" f"event={event}"
-                    )
-
-    def handle_token_uri_update(self, from_block, to_block):
-        events = self.get_event_logs(
-            EventTypes.EVENT_TOKEN_URI_UPDATE, from_block, to_block
-        )
+            erc20_contract = self._web3.eth.contract(
+                abi=ERC20Template.abi,
+                address=self._web3.toChecksumAddress(erc20_address),
+            )
+
+            logger.debug(f"{event_name} detected on ERC20 contract {event.address}.")
 
+            try:
+                event_processor = OrderStartedProcessor(
+                    erc20_contract.caller.getERC721Address(),
+                    self._es_instance,
+                    to_block,
+                    self._chain_id,
+                )
+                event_processor.process()
+            except Exception as e:
+                logger.error(
+                    f"Error processing {event_name} event: {e}\n" f"event={event}"
+                )
+
+    def handle_token_uri_update(self, events):
         for event in events:
             try:
                 event_processor = TokenURIUpdatedProcessor(
@@ -329,9 +365,7 @@ def handle_token_uri_update(self, from_block, to_block):
                     f"Error processing token update event: {e}\n" f"event={event}"
                 )
 
-    def handle_transfer_ownership(self, from_block, to_block):
-        events = self.get_event_logs(EventTypes.EVENT_TRANSFER, from_block, to_block)
-
+    def handle_transfer_ownership(self, events):
         for event in events:
             try:
                 event_processor = TransferProcessor(
@@ -507,3 +541,12 @@ def get_event_logs(self, event_name, from_block, to_block, chunk_size=1000):
         )
 
         return all_logs
+
+
+def merge_list_dictionary(dict_1, dict_2):
+    dict_3 = {**dict_1, **dict_2}
+    for key, value in dict_3.items():
+        if key in dict_1 and key in dict_2:
+            dict_3[key] = value + dict_1[key]
+
+    return dict_3