From 01387993b507a1803374888d90aca2bc7434a3e6 Mon Sep 17 00:00:00 2001 From: nolash Date: Mon, 22 Feb 2021 21:53:05 +0100 Subject: [PATCH] Make manager run with chainsyncer backend --- apps/cic-eth/cic_eth/registry.py | 86 +++++++++++ .../cic_eth/runnable/daemons/filters/gas.py | 3 +- .../cic_eth/runnable/daemons/manager.py | 138 +++--------------- apps/cic-eth/config/docker/database.ini | 2 +- 4 files changed, 108 insertions(+), 121 deletions(-) create mode 100644 apps/cic-eth/cic_eth/registry.py diff --git a/apps/cic-eth/cic_eth/registry.py b/apps/cic-eth/cic_eth/registry.py new file mode 100644 index 00000000..837e7f0f --- /dev/null +++ b/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 diff --git a/apps/cic-eth/cic_eth/runnable/daemons/filters/gas.py b/apps/cic-eth/cic_eth/runnable/daemons/filters/gas.py index 485f955e..3a44db8a 100644 --- a/apps/cic-eth/cic_eth/runnable/daemons/filters/gas.py +++ b/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): diff --git a/apps/cic-eth/cic_eth/runnable/daemons/manager.py b/apps/cic-eth/cic_eth/runnable/daemons/manager.py index 6aa11932..be4cab9e 100644 --- a/apps/cic-eth/cic_eth/runnable/daemons/manager.py +++ b/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.error import LoopDone +#from cic_eth.sync import Syncer +#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) - - #if config.get('ETH_ACCOUNT_ACCOUNTS_INDEX_WRITER') != None: - # CICRegistry.add_role(chain_spec, config.get('ETH_ACCOUNT_ACCOUNTS_INDEX_WRITER'), 'AccountRegistry', True) + registry = init_registry(config, w3) # Connect to blockchain with chainlib conn = HTTPConnection(config.get('ETH_PROVIDER')) @@ -169,38 +96,14 @@ def main(): logg.debug('starting at block {}'.format(block_offset)) - #syncers = [] - #block_offset = c.w3.eth.blockNumber - #chain = str(chain_spec) + syncers = [] - if SyncerBackend.first(config.get('CIC_CHAIN_SPEC')): - #from cic_eth.sync.history import HistorySyncer - backend = SyncerBackend.initial(config.get('CIC_CHAIN_SPEC'), block_offset) - #syncer = HistorySyncer(backend) - #syncers.append(syncer) + if SyncerBackend.first(chain_spec): + backend = SyncerBackend.initial(chain_spec, block_offset) - return - - #from cic_eth.sync.head import HeadSyncer - block_sync = SyncerBackend.live(config.get('CIC_CHAIN_SPEC'), block_offset+1) + block_sync = SyncerBackend.live(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: logg.critical('At least one trusted address must be declared in CIC_TRUST_ADDRESS') @@ -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.filter.append(registration_filter.filter) + syncer.add_filter(gas_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.filter.append(convert_filter) + syncer.add_filter(tx_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)) diff --git a/apps/cic-eth/config/docker/database.ini b/apps/cic-eth/config/docker/database.ini index 2b29289a..a5f2be48 100644 --- a/apps/cic-eth/config/docker/database.ini +++ b/apps/cic-eth/config/docker/database.ini @@ -6,4 +6,4 @@ HOST=localhost PORT=63432 ENGINE=postgresql DRIVER=psycopg2 -DEBUG=1 +DEBUG=0