Merge pull request #33 from bugout-dev/function-call-crawler-improvements

Function call crawler improvements

moonworm/crawler/function_call_crawler.py

@@ -2,7 +2,8 @@ import os
 import pickle
 from abc import ABC, abstractmethod
 from dataclasses import asdict, dataclass
-from typing import Any, Dict, List
+from logging import error
+from typing import Any, Callable, Dict, List, Optional
 
 from eth_typing.evm import ChecksumAddress
 from web3 import Web3
@@ -137,12 +138,14 @@ class FunctionCallCrawler:
         ethereum_state_provider: EthereumStateProvider,
         contract_abi: List[Dict[str, Any]],
         contract_addresses: List[ChecksumAddress],
+        on_decode_error: Optional[Callable[[Exception], None]] = None,
     ):
         self.state = state
         self.ethereum_state_provider = ethereum_state_provider
         self.contract_abi = contract_abi
         self.contract_addresses = contract_addresses
         self.contract = Web3().eth.contract(abi=self.contract_abi)
+        self.on_decode_error = on_decode_error
 
     def process_transaction(self, transaction: Dict[str, Any]):
         try:
@@ -175,6 +178,8 @@ class FunctionCallCrawler:
             self.state.register_call(function_call)
         except Exception as e:
             print(f"Failed to decode function call in tx: {transaction['hash'].hex()}")
+            if self.on_decode_error:
+                self.on_decode_error(e)
             print(e)
 
     def crawl(self, from_block: int, to_block: int, flush_state: bool = False):

moonworm/crawler/moonstream_ethereum_state_provider.py

@@ -28,15 +28,28 @@ class MoonstreamEthereumStateProvider(EthereumStateProvider):
     """
 
     def __init__(
-        self, w3: Web3, network: Network, db_session: Optional[Session] = None
+        self,
+        w3: Web3,
+        network: Network,
+        db_session: Optional[Session] = None,
+        batch_load_count: int = 100,
     ):
         self.w3 = w3
         self.db_session = db_session
 
+        self.metrics = {
+            "web3_get_block_calls": 0,
+            "web3_get_transaction_receipt_calls": 0,
+            "db_get_block_calls": 0,
+            "db_get_transaction_calls": 0,
+            "block_found_in_cache": 0,
+        }
+
         self.blocks_model = MODELS[network]["blocks"]
         self.transactions_model = MODELS[network]["transactions"]
         self.labels_model = MODELS[network]["labels"]
         self.network = network
+        self.batch_load_count = batch_load_count
         self.blocks_cache = {}
 
     def set_db_session(self, db_session: Session):
@@ -46,6 +59,7 @@ class MoonstreamEthereumStateProvider(EthereumStateProvider):
         self.db_session = None
 
     def get_transaction_reciept(self, transaction_hash: str) -> Dict[str, Any]:
+        self.metrics["web3_get_transaction_receipt_calls"] += 1
         return self.w3.eth.get_transaction_receipt(transaction_hash)
 
     def get_last_block_number(self) -> int:
@@ -81,17 +95,17 @@ class MoonstreamEthereumStateProvider(EthereumStateProvider):
         }
         return tx
 
-    def _get_block_from_db(
-        self, block_number: int, batch_load_count: int = 100
-    ) -> Optional[Dict[str, Any]]:
+    def _get_block_from_db(self, block_number: int) -> Optional[Dict[str, Any]]:
         if self.db_session is None:
             return None
 
+        self.metrics["db_get_block_calls"] += 1
+
         raw_blocks = (
             self.db_session.query(self.blocks_model)
             .filter(self.blocks_model.block_number >= block_number)
             .order_by(self.blocks_model.block_number.asc())
-            .limit(batch_load_count)
+            .limit(self.batch_load_count)
         )
         blocks = {raw_block.block_number: raw_block for raw_block in raw_blocks}
 
@@ -108,6 +122,7 @@ class MoonstreamEthereumStateProvider(EthereumStateProvider):
             .order_by(self.transactions_model.transaction_index.asc())
             .all()
         )
+        self.metrics["db_get_transaction_calls"] += 1
 
         block_transactions = {}
 
@@ -135,6 +150,7 @@ class MoonstreamEthereumStateProvider(EthereumStateProvider):
         logger.debug(log_prefix)
         if block_number in self.blocks_cache:
             logger.debug(f"{log_prefix} - found in cache")
+            self.metrics["block_found_in_cache"] += 1
             return self.blocks_cache[block_number]
 
         block = self._get_block_from_db(block_number)

moonworm/version.py

@@ -1 +1 @@
-MOONWORM_VERSION = "0.1.6"
+MOONWORM_VERSION = "0.1.7"