chainsyncer

Blockchain syncer driver
Log | Files | Refs | LICENSE

commit e8370de015cb906c03e3c831fdeea3590b322303
Author: nolash <dev@holbrook.no>
Date:   Wed,  3 Feb 2021 19:40:03 +0100

Initial commit

Diffstat:
Acic_syncer/client.py | 45+++++++++++++++++++++++++++++++++++++++++++++
Acic_syncer/runnable/tracker.py | 133+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Asql/postgresql/1.sql | 12++++++++++++
3 files changed, 190 insertions(+), 0 deletions(-)

diff --git a/cic_syncer/client.py b/cic_syncer/client.py @@ -0,0 +1,45 @@ +# standard imports +import uuid +import logging + +# third-party imports +import websockets + +logg = logging.getLogger() + + +class Syncer: + + def __init__(self, backend): + super(HeadSyncer, self).__init__(backend) + + +class MinedSyncer(Syncer): + + def __init__(self, backend): + super(HeadSyncer, self).__init__(backend) + + + +class HeadSyncer(MinedSyncer): + + def __init__(self, backend): + super(HeadSyncer, self).__init__(backend) + + + def get(self, getter): + (block_number, tx_number) = self.backend + block_hash = [] + try: + uu = uuid.uuid4() + req = { + 'jsonrpc': '2.0', + 'method': 'eth_getBlock', + 'id': str(uu), + 'param': [block_number], + logg.debug(req) + except Exception as e: + logg.error(e) + + return block_hash + diff --git a/cic_syncer/runnable/tracker.py b/cic_syncer/runnable/tracker.py @@ -0,0 +1,133 @@ +# standard imports +import os +import sys +import logging +import time +import argparse +import sys +import re + +# third-party imports +import confini +import rlp + + +# local imports + + +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)) + +app = celery.Celery(backend=config.get('CELERY_RESULT_URL'), broker=config.get('CELERY_BROKER_URL')) + +queue = args.q + +dsn = dsn_from_config(config) +SessionBase.connect(dsn) + +# TODO: There is too much code in this file, split it up + +transfer_callbacks = [] +for cb in config.get('TASKS_SYNCER_CALLBACKS', '').split(','): + task_split = cb.split(':') + task_queue = queue + if len(task_split) > 1: + task_queue = task_split[0] + task_pair = (task_split[1], task_queue) + transfer_callbacks.append(task_pair) + + +def tx_filter(w3, tx, rcpt, chain_spec): + tx_hash_hex = tx.hash.hex() + otx = Otx.load(tx_hash_hex) + if otx == None: + logg.debug('tx {} not found locally, skipping'.format(tx_hash_hex)) + return None + logg.info('otx found {}'.format(otx.tx_hash)) + s = celery.signature( + 'cic_eth.queue.tx.set_final_status', + [ + tx_hash_hex, + rcpt.blockNumber, + rcpt.status == 0, + ], + queue=queue, + ) + t = s.apply_async() + return t + + + +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 main(): + chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC')) + c = RpcClient(chain_spec) + + block_offset = c.w3.eth.blockNumber + chain = str(chain_spec) + + for cb in config.get('TASKS_SYNCER_CALLBACKS', '').split(','): + task_split = cb.split(':') + task_queue = queue + if len(task_split) > 1: + task_queue = task_split[0] + task_pair = (task_split[1], task_queue) + syncer.filter.append(task_pair) + + try: + syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL'))) + except LoopDone as e: + sys.stderr.write("sync '{}' done at block {}\n".format(args.mode, e)) + + sys.exit(0) + + +if __name__ == '__main__': + main() diff --git a/sql/postgresql/1.sql b/sql/postgresql/1.sql @@ -0,0 +1,12 @@ +DROP TABLE chain_sync; +CREATE TABLE IF NOT EXISTS chain_sync ( + id serial primary key not null, + blockchain varchar not null, + block_start int not null default 0, + tx_start int not null default 0, + block_cursor int not null default 0, + tx_cursor int not null default 0, + block_target int default null, + date_created timestamp not null, + date_updated timestamp default null +);