Fix inspection issues

proxy/indexer/indexer.py

@@ -1,18 +1,22 @@
-from proxy.indexer.indexer_base import logger, IndexerBase
+from proxy.indexer.indexer_base import logger, IndexerBase, PARALLEL_REQUESTS
 import base58
 import rlp
 import json
 import os
+import time
 import logging
+from multiprocessing.dummy import Pool as ThreadPool
+
 
 try:
-    from utils import check_error, get_trx_results, get_trx_receipts, LogDB
+    from utils import check_error, get_trx_results, get_trx_receipts, LogDB, Canceller
     from sql_dict import SQLDict
 except ImportError:
-    from .utils import check_error, get_trx_results, get_trx_receipts, LogDB
+    from .utils import check_error, get_trx_results, get_trx_receipts, LogDB, Canceller
     from .sql_dict import SQLDict
 
 CANCEL_TIMEOUT = int(os.environ.get("CANCEL_TIMEOUT", "60"))
+UPDATE_BLOCK_COUNT = PARALLEL_REQUESTS * 16
 
 class HolderStruct:
     def __init__(self, storage_account):
@@ -48,17 +52,36 @@ def __init__(self,
                  evm_loader_id,
                  log_level = 'INFO'):
         IndexerBase.__init__(self, solana_url, evm_loader_id, log_level)
+
+        self.canceller = Canceller()
         self.logs_db = LogDB()
+        self.blocks_by_hash = SQLDict(tablename="solana_blocks_by_hash")
         self.ethereum_trx = SQLDict(tablename="ethereum_transactions")
         self.eth_sol_trx = SQLDict(tablename="ethereum_solana_transactions")
         self.sol_eth_trx = SQLDict(tablename="solana_ethereum_transactions")
+        self.constants = SQLDict(tablename="constants")
+        if 'last_block' not in self.constants:
+            self.constants['last_block'] = 0
+        self.blocked_storages = {}
+
+
+    def process_functions(self):
+        IndexerBase.process_functions(self)
+        logger.debug("Process receipts")
+        self.process_receipts()
+        logger.debug("Start getting blocks")
+        self.gather_blocks()
+        logger.debug("Unlock accounts")
+        self.canceller.unlock_accounts(self.blocked_storages)
+        self.blocked_storages = {}
 
 
     def process_receipts(self):
         counter = 0
         holder_table = {}
         continue_table = {}
         trx_table = {}
+
         for signature in self.transaction_order:
             counter += 1
 
@@ -410,6 +433,39 @@ def submit_transaction(self, trx_struct):
         logger.debug(trx_struct.eth_signature + " " + status)
 
 
+    def gather_blocks(self):
+        max_slot = self.client.get_slot(commitment="recent")["result"]
+
+        last_block = self.constants['last_block']
+        if last_block + UPDATE_BLOCK_COUNT < max_slot:
+            max_slot = last_block + UPDATE_BLOCK_COUNT
+        slots = self.client._provider.make_request("getBlocks", last_block, max_slot, {"commitment": "confirmed"})["result"]
+
+        pool = ThreadPool(PARALLEL_REQUESTS)
+        results = pool.map(self.get_block, slots)
+
+        for block_result in results:
+            (slot, block_hash) = block_result
+            self.blocks_by_hash[block_hash] = slot
+
+        self.constants['last_block'] = max_slot
+
+
+    def get_block(self, slot):
+        retry = True
+
+        while retry:
+            try:
+                block = self.client._provider.make_request("getBlock", slot, {"commitment":"confirmed", "transactionDetails":"none", "rewards":False})['result']
+                block_hash = '0x' + base58.b58decode(block['blockhash']).hex()
+                retry = False
+            except Exception as err:
+                logger.debug(err)
+                time.sleep(1)
+
+        return (slot, block_hash)
+
+
 def run_indexer(solana_url,
                 evm_loader_id,
                 log_level = 'INFO'):

proxy/indexer/indexer_base.py

@@ -1,4 +1,3 @@
-import base58
 import os
 import time
 import logging
@@ -7,10 +6,8 @@
 from typing import Dict, Union
 
 try:
-    from utils import Canceller
     from sql_dict import SQLDict
 except ImportError:
-    from .utils import Canceller
     from .sql_dict import SQLDict
 
 
@@ -22,7 +19,6 @@
 DEVNET_HISTORY_START = "7BdwyUQ61RUZP63HABJkbW66beLk22tdXnP69KsvQBJekCPVaHoJY47Rw68b3VV1UbQNHxX3uxUSLfiJrfy2bTn"
 HISTORY_START = [DEVNET_HISTORY_START]
 
-UPDATE_BLOCK_COUNT = PARALLEL_REQUESTS * 16
 
 log_levels = {
     'DEBUG': logging.DEBUG,
@@ -39,38 +35,28 @@ def __init__(self,
                  solana_url,
                  evm_loader_id,
                  log_level):
+        logger.setLevel(log_levels.get(log_level, logging.INFO))
+
         self.evm_loader_id = evm_loader_id
         self.client = Client(solana_url)
-        self.canceller = Canceller()
-        self.blocks_by_hash = SQLDict(tablename="solana_blocks_by_hash")
         self.transaction_receipts = SQLDict(tablename="known_transactions")
-        self.constants = SQLDict(tablename="constants")
-
-        logger.setLevel(log_levels.get(log_level, logging.INFO))
-
         self.last_slot = 0
         self.current_slot = 0
         self.transaction_order = []
-        if 'last_block' not in self.constants:
-            self.constants['last_block'] = 0
-        self.blocked_storages = {}
         self.counter_ = 0
 
 
     def run(self):
         while (True):
             try:
-                logger.debug("Start indexing")
-                self.gather_unknown_transactions()
-                logger.debug("Process receipts")
-                self.process_receipts()
-                logger.debug("Start getting blocks")
-                self.gather_blocks()
-                logger.debug("Unlock accounts")
-                self.canceller.unlock_accounts(self.blocked_storages)
-                self.blocked_storages = {}
+                self.process_functions()
             except Exception as err:
-                logger.debug("Got exception while indexing. Type(err):%s, Exception:%s", type(err), err)
+                logger.warning("Got exception while indexing. Type(err):%s, Exception:%s", type(err), err)
+
+
+    def process_functions(self):
+        logger.debug("Start indexing")
+        self.gather_unknown_transactions()
 
 
     def gather_unknown_transactions(self):
@@ -161,42 +147,3 @@ def get_tx_receipts(self, solana_signature):
             logger.debug(self.counter_)
 
         # return (solana_signature, trx)
-
-
-    def process_receipts(self): raise Exception('NotImplemented')
-
-    def gather_blocks(self):
-        max_slot = self.client.get_slot(commitment="recent")["result"]
-
-        last_block = self.constants['last_block']
-        if last_block + UPDATE_BLOCK_COUNT < max_slot:
-            max_slot = last_block + UPDATE_BLOCK_COUNT
-        slots = self.client._provider.make_request("getBlocks", last_block, max_slot, {"commitment": "confirmed"})["result"]
-
-        pool = ThreadPool(PARALLEL_REQUESTS)
-        results = pool.map(self.get_block, slots)
-
-        for block_result in results:
-            (slot, block_hash) = block_result
-            self.blocks_by_hash[block_hash] = slot
-
-        self.constants['last_block'] = max_slot
-
-
-    def get_block(self, slot):
-        retry = True
-
-        while retry:
-            try:
-                block = self.client._provider.make_request("getBlock", slot, {"commitment":"confirmed", "transactionDetails":"none", "rewards":False})['result']
-                block_hash = '0x' + base58.b58decode(block['blockhash']).hex()
-                retry = False
-            except Exception as err:
-                logger.debug(err)
-                time.sleep(1)
-
-        return (slot, block_hash)
-
-
-
-