Make manager run with chainsyncer backend

apps/cic-eth/cic_eth/registry.py

@@ -0,0 +1,86 @@
+# standard imports
+import logging
+import copy
+
+# external imports
+from cic_registry import CICRegistry
+from eth_token_index import TokenUniqueSymbolIndex
+from eth_accounts_index import AccountRegistry
+from chainlib.chain import ChainSpec
+from cic_registry.chain import ChainRegistry
+from cic_registry.helper.declarator import DeclaratorOracleAdapter
+
+logg = logging.getLogger(__name__)
+
+
+class TokenOracle:
+
+    def __init__(self, conn, chain_spec, registry):
+        self.tokens = []
+        self.chain_spec = chain_spec
+        self.registry = registry
+
+        token_registry_contract = CICRegistry.get_contract(chain_spec, 'TokenRegistry', 'Registry')
+        self.getter = TokenUniqueSymbolIndex(conn, token_registry_contract.address())
+
+
+    def get_tokens(self):
+        token_count = self.getter.count()
+        if token_count == len(self.tokens):
+            return self.tokens
+
+        for i in range(len(self.tokens), token_count):
+            token_address = self.getter.get_index(i)
+            t = self.registry.get_address(self.chain_spec, token_address)
+            token_symbol = t.symbol()
+            self.tokens.append(t)
+
+            logg.debug('adding token idx {} symbol {} address {}'.format(i, token_symbol, token_address))
+
+        return copy.copy(self.tokens)
+
+
+class AccountsOracle:
+
+    def __init__(self, conn, chain_spec, registry):
+        self.accounts = []
+        self.chain_spec = chain_spec
+        self.registry = registry
+
+        accounts_registry_contract = CICRegistry.get_contract(chain_spec, 'AccountRegistry', 'Registry')
+        self.getter = AccountRegistry(conn, accounts_registry_contract.address())
+
+
+    def get_accounts(self):
+        accounts_count = self.getter.count()
+        if accounts_count == len(self.accounts):
+            return self.accounts
+
+        for i in range(len(self.accounts), accounts_count):
+            account = self.getter.get_index(i)
+            self.accounts.append(account)
+            logg.debug('adding account {}'.format(account))
+
+        return copy.copy(self.accounts)
+
+
+def init_registry(config, w3):
+    chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
+    CICRegistry.init(w3, config.get('CIC_REGISTRY_ADDRESS'), chain_spec)
+    CICRegistry.add_path(config.get('ETH_ABI_DIR'))
+
+    chain_registry = ChainRegistry(chain_spec)
+    CICRegistry.add_chain_registry(chain_registry, True)
+
+    declarator = CICRegistry.get_contract(chain_spec, 'AddressDeclarator', interface='Declarator')
+    trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
+    if trusted_addresses_src == None:
+        raise ValueError('At least one trusted address must be declared in CIC_TRUST_ADDRESS')
+    trusted_addresses = trusted_addresses_src.split(',')
+    for address in trusted_addresses:
+        logg.info('using trusted address {}'.format(address))
+
+    oracle = DeclaratorOracleAdapter(declarator.contract, trusted_addresses)
+    chain_registry.add_oracle(oracle, 'naive_erc20_oracle')
+
+    return CICRegistry

apps/cic-eth/cic_eth/runnable/daemons/filters/gas.py

@@ -18,9 +18,8 @@ logg = logging.getLogger()
 
 class GasFilter(SyncFilter):
 
-    def __init__(self, gas_provider, queue=None):
+    def __init__(self, queue=None):
         self.queue = queue
-        self.gas_provider = gas_provider
 
 
     def filter(self, w3, tx, rcpt, chain_str, session=None):

apps/cic-eth/cic_eth/runnable/daemons/manager.py

@@ -13,12 +13,12 @@ import celery
 import rlp
 import web3
 from web3 import HTTPProvider, WebsocketProvider
+import cic_base
 from cic_registry import CICRegistry
 from chainlib.chain import ChainSpec
 from cic_registry import zero_address
 from cic_registry.chain import ChainRegistry
 from cic_registry.error import UnknownContractError
-#from cic_bancor.bancor import BancorRegistryClient
 from chainlib.eth.connection import HTTPConnection
 from chainlib.eth.block import (
         block_latest,
@@ -27,21 +27,21 @@ from hexathon import (
         strip_0x,
         )
 from chainsyncer.backend import SyncerBackend
+from chainsyncer.driver import HeadSyncer
 from chainsyncer.db.models.base import SessionBase
-import cic_base
+from chainsyncer.error import LoopDone
 
 # local imports
-import cic_eth
+from cic_eth.registry import init_registry
 from cic_eth.eth import RpcClient
-#from cic_eth.db import SessionBase
 from cic_eth.db import Otx
 from cic_eth.db import TxConvertTransfer
 from cic_eth.db.models.tx import TxCache
 from cic_eth.db.enum import StatusEnum
 from cic_eth.db import dsn_from_config
 from cic_eth.queue.tx import get_paused_txs
-from cic_eth.sync import Syncer
+#from cic_eth.sync import Syncer
-from cic_eth.sync.error import LoopDone
+#from cic_eth.sync.error import LoopDone
 from cic_eth.db.error import UnknownConvertError
 from cic_eth.eth.util import unpack_signed_raw_tx
 from cic_eth.eth.task import create_check_gas_and_send_task
@@ -55,8 +55,6 @@ from cic_eth.runnable.daemons.filters import (
         RegistrationFilter,
         )
 
-
-
 script_dir = os.path.realpath(os.path.dirname(__file__))
 
 logg = cic_base.log.create()
@@ -65,80 +63,12 @@ argparser = cic_base.argparse.create(script_dir, cic_base.argparse.full_template
 args = cic_base.argparse.parse(argparser, logg)
 config = cic_base.config.create(args.c, args, args.env_prefix)
 
-#logging.basicConfig(level=logging.WARNING)
-#logg = logging.getLogger()
-#logging.getLogger('websockets.protocol').setLevel(logging.CRITICAL)
-#logging.getLogger('web3.RequestManager').setLevel(logging.CRITICAL)
-#logging.getLogger('web3.providers.WebsocketProvider').setLevel(logging.CRITICAL)
-#logging.getLogger('web3.providers.HTTPProvider').setLevel(logging.CRITICAL)
-
-#config_dir = os.path.join('/usr/local/etc/cic-eth')
-#
-#argparser = argparse.ArgumentParser(description='daemon that monitors transactions in new blocks')
-#argparser.add_argument('-c', type=str, default=config_dir, help='config root to use')
-#argparser.add_argument('-i', '--chain-spec', type=str, dest='i', help='chain spec')
-#argparser.add_argument('--abi-dir', dest='abi_dir', type=str, help='Directory containing bytecode and abi')
-#argparser.add_argument('--env-prefix', default=os.environ.get('CONFINI_ENV_PREFIX'), dest='env_prefix', type=str, help='environment prefix for variables to overwrite configuration')
-#argparser.add_argument('-q', type=str, default='cic-eth', help='celery queue to submit transaction tasks to')
-#argparser.add_argument('-v', help='be verbose', action='store_true')
-#argparser.add_argument('-vv', help='be more verbose', action='store_true')
-#argparser.add_argument('mode', type=str, help='sync mode: (head|history)', default='head')
-#args = argparser.parse_args(sys.argv[1:])
-#
-#if args.v == True:
-#    logging.getLogger().setLevel(logging.INFO)
-#elif args.vv == True:
-#    logging.getLogger().setLevel(logging.DEBUG)
-#
-#config_dir = os.path.join(args.c)
-#os.makedirs(config_dir, 0o777, True)
-#config = confini.Config(config_dir, args.env_prefix)
-#config.process()
-## override args
-#args_override = {
-#        'ETH_ABI_DIR': getattr(args, 'abi_dir'),
-#        'CIC_CHAIN_SPEC': getattr(args, 'i'),
-#        }
-#config.dict_override(args_override, 'cli flag')
-#config.censor('PASSWORD', 'DATABASE')
-#config.censor('PASSWORD', 'SSL')
-#logg.debug('config loaded from {}:\n{}'.format(config_dir, config))
-
 config.add(args.y, '_KEYSTORE_FILE', True)
 
 config.add(args.q, '_CELERY_QUEUE', True)
 
 cic_base.config.log(config)
 
-#app = celery.Celery(backend=config.get('CELERY_RESULT_URL'),  broker=config.get('CELERY_BROKER_URL'))
-
-#queue = args.q
-#
-#chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
-
-
-#re_websocket = re.compile('^wss?://')
-#re_http = re.compile('^https?://')
-#blockchain_provider = config.get('ETH_PROVIDER')
-#if re.match(re_websocket, blockchain_provider) != None:
-#    blockchain_provider = WebsocketProvider(blockchain_provider)
-#elif re.match(re_http, blockchain_provider) != None:
-#    blockchain_provider = HTTPProvider(blockchain_provider)
-#else:
-#    raise ValueError('unknown provider url {}'.format(blockchain_provider))
-#
-#def web3_constructor():
-#    w3 = web3.Web3(blockchain_provider)
-#    return (blockchain_provider, w3)
-#RpcClient.set_constructor(web3_constructor)
-#
-#c = RpcClient(chain_spec)
-#CICRegistry.init(c.w3, config.get('CIC_REGISTRY_ADDRESS'), chain_spec)
-#CICRegistry.add_path(config.get('ETH_ABI_DIR'))
-#chain_registry = ChainRegistry(chain_spec)
-#CICRegistry.add_chain_registry(chain_registry, True)
-
-
 
 dsn = dsn_from_config(config)
 SessionBase.connect(dsn, pool_size=1, debug=config.true('DATABASE_DEBUG'))
@@ -155,10 +85,7 @@ def main():
 
     # set up registry
     w3 = cic_base.rpc.create(config.get('ETH_PROVIDER')) # replace with HTTPConnection when registry has been so refactored
-    registry = cic_base.registry.init_legacy(config, w3)
+    registry = init_registry(config, w3)
-
-    #if config.get('ETH_ACCOUNT_ACCOUNTS_INDEX_WRITER') != None:
-    #    CICRegistry.add_role(chain_spec, config.get('ETH_ACCOUNT_ACCOUNTS_INDEX_WRITER'), 'AccountRegistry', True)
 
     # Connect to blockchain with chainlib
     conn = HTTPConnection(config.get('ETH_PROVIDER'))
@@ -169,37 +96,13 @@ def main():
 
     logg.debug('starting at block {}'.format(block_offset))
 
-    #syncers = []
+    syncers = []
-    #block_offset = c.w3.eth.blockNumber
-    #chain = str(chain_spec)
 
-    if SyncerBackend.first(config.get('CIC_CHAIN_SPEC')):
+    if SyncerBackend.first(chain_spec):
-        #from cic_eth.sync.history import HistorySyncer
+        backend = SyncerBackend.initial(chain_spec, block_offset)
-        backend = SyncerBackend.initial(config.get('CIC_CHAIN_SPEC'), block_offset)
-        #syncer = HistorySyncer(backend) 
-        #syncers.append(syncer)
 
-    return
+    block_sync = SyncerBackend.live(chain_spec, block_offset+1)
-
-    #from cic_eth.sync.head import HeadSyncer
-    block_sync = SyncerBackend.live(config.get('CIC_CHAIN_SPEC'), block_offset+1)
     syncers.append(HeadSyncer(block_sync))
-#    elif args.mode == 'history':
-#        from cic_eth.sync.history import HistorySyncer
-#        backends = SyncerBackend.resume(chain, block_offset+1)
-#        for backend in backends:
-#            syncers.append(HistorySyncer(backend))
-#        if len(syncers) == 0:
-#            logg.info('found no unsynced history. terminating')
-#            sys.exit(0)
-#    else:
-#        sys.stderr.write("unknown mode '{}'\n".format(args.mode))
-#        sys.exit(1)
-
-#    bancor_registry_contract = CICRegistry.get_contract(chain_spec, 'BancorRegistry', interface='Registry')
-#    bancor_chain_registry = CICRegistry.get_chain_registry(chain_spec)
-#    bancor_registry = BancorRegistryClient(c.w3, bancor_chain_registry, config.get('ETH_ABI_DIR'))
-#    bancor_registry.load() 
 
     trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
     if trusted_addresses_src == None:
@@ -213,31 +116,30 @@ def main():
     callback_filters = []
     for cb in config.get('TASKS_TRANSFER_CALLBACKS', '').split(','):
         task_split = cb.split(':')
-        task_queue = queue
+        task_queue = config.get('_CELERY_QUEUE')
         if len(task_split) > 1:
             task_queue = task_split[0]
         callback_filter = CallbackFilter(task_split[1], task_queue)
         callback_filters.append(callback_filter)
 
-    tx_filter = TxFilter(queue)
+    tx_filter = TxFilter(config.get('_CELERY_QUEUE'))
 
-    registration_filter = RegistrationFilter(queue)
+    registration_filter = RegistrationFilter(config.get('_CELERY_QUEUE'))
 
-    gas_filter = GasFilter(c.gas_provider(), queue)
+    gas_filter = GasFilter(config.get('_CELERY_QUEUE'))
 
     i = 0
     for syncer in syncers:
         logg.debug('running syncer index {}'.format(i))
-        syncer.filter.append(gas_filter.filter)
+        syncer.add_filter(gas_filter)
-        syncer.filter.append(registration_filter.filter)
+        syncer.add_filter(registration_filter)
         # TODO: the two following filter functions break the filter loop if return uuid. Pro: less code executed. Con: Possibly unintuitive flow break
-        syncer.filter.append(tx_filter.filter)
+        syncer.add_filter(tx_filter)
-        #syncer.filter.append(convert_filter)
         for cf in callback_filters:
-            syncer.filter.append(cf.filter)
+            syncer.add_filter(cf)
 
         try:
-            syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')))
+            syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')), conn)
         except LoopDone as e:
             sys.stderr.write("sync '{}' done at block {}\n".format(args.mode, e))
 

apps/cic-eth/config/docker/database.ini

@@ -6,4 +6,4 @@ HOST=localhost
 PORT=63432
 ENGINE=postgresql
 DRIVER=psycopg2
-DEBUG=1
+DEBUG=0