Merge remote-tracking branch 'origin/master' into lash/rehabilitate-cic-cache

Rehabilitate retrier

See merge request grassrootseconomics/cic-internal-integration!77

apps/cic-cache/cic_cache/db/migrations/default/versions/6604de4203e2_add_chain_syncer.py

@@ -0,0 +1,28 @@
+"""Add chain syncer
+
+Revision ID: 6604de4203e2
+Revises: 63b629f14a85
+Create Date: 2021-04-01 08:10:29.156243
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from chainsyncer.db.migrations.sqlalchemy import (
+        chainsyncer_upgrade,
+        chainsyncer_downgrade,
+        )
+
+
+# revision identifiers, used by Alembic.
+revision = '6604de4203e2'
+down_revision = '63b629f14a85'
+branch_labels = None
+depends_on = None
+
+def upgrade():
+    chainsyncer_upgrade(0, 0, 1)
+
+
+def downgrade():
+    chainsyncer_downgrade(0, 0, 1)
+

apps/cic-cache/cic_cache/runnable/daemons/filters/__init__.py

@@ -0,0 +1 @@
+from .erc20 import *

apps/cic-cache/cic_cache/runnable/daemons/filters/base.py

@@ -0,0 +1,2 @@
+class SyncFilter:
+    pass

apps/cic-cache/cic_cache/runnable/daemons/filters/erc20.py

@@ -0,0 +1,72 @@
+# standard imports
+import logging
+
+# external imports
+from chainlib.eth.erc20 import ERC20
+from chainlib.eth.address import (
+        to_checksum_address,
+        )
+from chainlib.eth.error import RequestMismatchException
+from chainlib.status import Status
+from cic_eth_registry.erc20 import ERC20Token
+from cic_eth_registry.error import (
+        NotAContractError,
+        ContractMismatchError,
+        )
+
+# local imports
+from .base import SyncFilter
+from cic_cache import db as cic_cache_db
+
+logg = logging.getLogger().getChild(__name__)
+
+
+class ERC20TransferFilter(SyncFilter):
+
+    def __init__(self, chain_spec):
+        self.chain_spec = chain_spec
+
+
+    # TODO: Verify token in declarator / token index
+    def filter(self, conn, block, tx, db_session=None):
+        logg.debug('filter {} {}'.format(block, tx))
+        token = None
+        try:
+            token = ERC20Token(conn, tx.inputs[0])
+        except NotAContractError:
+            logg.debug('not a contract {}'.format(tx.inputs[0]))
+            return False
+        except ContractMismatchError:
+            logg.debug('not an erc20 token  {}'.format(tx.inputs[0]))
+            return False
+
+        transfer_data = None
+        try:
+            transfer_data = ERC20.parse_transfer_request(tx.payload)
+        except RequestMismatchException:
+            logg.debug('erc20 match but not a transfer, skipping')
+            return False
+
+        token_sender = tx.outputs[0]
+        token_recipient = transfer_data[0]
+        token_value = transfer_data[1]
+
+        logg.debug('matched erc20 token transfer {} ({}) to {} value {}'.format(token.name, token.address, transfer_data[0], transfer_data[1]))
+
+        cic_cache_db.add_transaction(
+                db_session,
+                tx.hash,
+                block.number,
+                tx.index,
+                to_checksum_address(token_sender),
+                to_checksum_address(token_recipient),
+                token.address,
+                token.address,
+                token_value,
+                token_value,
+                tx.status == Status.SUCCESS,
+                block.timestamp,
+                )
+        db_session.flush()
+
+        return True

apps/cic-cache/cic_cache/runnable/daemons/tracker.py

@@ -0,0 +1,109 @@
+# standard imports
+import os
+import sys
+import logging
+import time
+import argparse
+import sys
+import re
+
+# third-party imports
+import confini
+import celery
+import rlp
+import cic_base.config
+import cic_base.log
+import cic_base.argparse
+import cic_base.rpc
+from cic_eth_registry import CICRegistry
+from cic_eth_registry.error import UnknownContractError
+from chainlib.chain import ChainSpec
+from chainlib.eth.constant import ZERO_ADDRESS
+from chainlib.connection import RPCConnection
+from chainlib.eth.block import (
+        block_latest,
+        )
+from hexathon import (
+        strip_0x,
+        )
+from chainsyncer.backend import SyncerBackend
+from chainsyncer.driver import (
+        HeadSyncer,
+        )
+from chainsyncer.db.models.base import SessionBase
+
+# local imports
+from cic_cache.db import dsn_from_config
+from cic_cache.runnable.daemons.filters import (
+        ERC20TransferFilter,
+        )
+
+script_dir = os.path.realpath(os.path.dirname(__file__))
+
+logg = cic_base.log.create()
+argparser = cic_base.argparse.create(script_dir, cic_base.argparse.full_template)
+#argparser = cic_base.argparse.add(argparser, add_traffic_args, 'traffic')
+args = cic_base.argparse.parse(argparser, logg)
+config = cic_base.config.create(args.c, args, args.env_prefix)
+
+cic_base.config.log(config)
+
+dsn = dsn_from_config(config)
+
+SessionBase.connect(dsn, debug=config.true('DATABASE_DEBUG'))
+
+chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
+
+#RPCConnection.register_location(config.get('ETH_PROVIDER'), chain_spec, 'default')
+cic_base.rpc.setup(chain_spec, config.get('ETH_PROVIDER'))
+
+
+def main():
+    # Connect to blockchain with chainlib
+    rpc = RPCConnection.connect(chain_spec, 'default')
+
+    o = block_latest()
+    r = rpc.do(o)
+    block_offset = int(strip_0x(r), 16) + 1
+
+    logg.debug('starting at block {}'.format(block_offset))
+
+    syncers = []
+
+    #if SyncerBackend.first(chain_spec):
+    #    backend = SyncerBackend.initial(chain_spec, block_offset)
+    syncer_backends = SyncerBackend.resume(chain_spec, block_offset)
+
+    if len(syncer_backends) == 0:
+        logg.info('found no backends to resume')
+        syncer_backends.append(SyncerBackend.initial(chain_spec, block_offset))
+    else:
+        for syncer_backend in syncer_backends:
+            logg.info('resuming sync session {}'.format(syncer_backend))
+
+    syncer_backends.append(SyncerBackend.live(chain_spec, block_offset+1))
+
+    syncers.append(HeadSyncer(syncer_backend))
+
+    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')
+        sys.exit(1)
+    trusted_addresses = trusted_addresses_src.split(',')
+    for address in trusted_addresses:
+        logg.info('using trusted address {}'.format(address))
+
+    erc20_transfer_filter = ERC20TransferFilter(chain_spec)
+
+    i = 0
+    for syncer in syncers:
+        logg.debug('running syncer index {}'.format(i))
+        syncer.add_filter(erc20_transfer_filter)
+        r = syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')), rpc)
+        sys.stderr.write("sync {} done at block {}\n".format(syncer, r))
+
+        i += 1
+
+
+if __name__ == '__main__':
+    main()

apps/cic-cache/cic_cache/runnable/tracker.py

@@ -1,339 +0,0 @@
-# standard imports
-import sys
-import os
-import argparse
-import logging
-import time
-import enum
-import re
-
-# third-party imports
-import confini
-from cic_registry import CICRegistry
-from cic_registry.chain import (
-        ChainRegistry,
-        ChainSpec,
-        )
-#from cic_registry.bancor import BancorRegistryClient
-from cic_registry.token import Token
-from cic_registry.error import (
-        UnknownContractError,
-        UnknownDeclarationError,
-        )
-from cic_registry.declaration import to_token_declaration
-from web3.exceptions import BlockNotFound, TransactionNotFound
-from websockets.exceptions import ConnectionClosedError
-from requests.exceptions import ConnectionError
-import web3
-from web3 import HTTPProvider, WebsocketProvider
-
-# local imports
-from cic_cache import db
-from cic_cache.db.models.base import SessionBase
-
-logging.basicConfig(level=logging.WARNING)
-logg = logging.getLogger()
-logging.getLogger('websockets.protocol').setLevel(logging.CRITICAL)
-logging.getLogger('urllib3').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)
-
-log_topics = {
-    'transfer': '0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef',
-    'convert': '0x7154b38b5dd31bb3122436a96d4e09aba5b323ae1fd580025fab55074334c095',
-    'accountregistry_add': '0a3b0a4f4c6e53dce3dbcad5614cb2ba3a0fa7326d03c5d64b4fa2d565492737',
-    }
-
-config_dir = os.path.join('/usr/local/etc/cic-cache')
-
-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('--trust-address', default=[], type=str, dest='trust_address', action='append', help='Set address as trust')
-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('--abi-dir', dest='abi_dir', type=str, help='Directory containing bytecode and abi')
-argparser.add_argument('-v', help='be verbose', action='store_true')
-argparser.add_argument('-vv', help='be more verbose', action='store_true')
-args = argparser.parse_args(sys.argv[1:])
-
-config_dir = os.path.join(args.c)
-os.makedirs(config_dir, 0o777, True)
-
-
-if args.v == True:
-    logging.getLogger().setLevel(logging.INFO)
-elif args.vv == True:
-    logging.getLogger().setLevel(logging.DEBUG)
-
-config = confini.Config(config_dir, args.env_prefix)
-config.process()
-args_override = {
-        'ETH_ABI_DIR': getattr(args, 'abi_dir'),
-        'CIC_TRUST_ADDRESS': ",".join(getattr(args, 'trust_address', [])),
-        }
-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))
-
-# connect to database
-dsn = db.dsn_from_config(config)
-SessionBase.connect(dsn)
-
-
-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)
-
-
-class RunStateEnum(enum.IntEnum):
-    INIT = 0
-    RUN = 1
-    TERMINATE = 9
-
-
-def rubberstamp(src):
-    return True
-
-
-class Tracker:
-
-    def __init__(self, chain_spec, trusts=[]):
-        self.block_height = 0
-        self.tx_height = 0
-        self.state = RunStateEnum.INIT
-        self.declarator_cache = {}
-        self.convert_enabled = False
-        self.trusts = trusts
-        self.chain_spec = chain_spec
-        self.declarator = CICRegistry.get_contract(chain_spec, 'AddressDeclarator', 'Declarator')
-
-
-    def __process_tx(self, w3, session, t, r, l, b):
-        token_value = int(l.data, 16)
-        token_sender = l.topics[1][-20:].hex()
-        token_recipient = l.topics[2][-20:].hex()
-
-        #ts = ContractRegistry.get_address(t.address)
-        ts = CICRegistry.get_address(self.chain_spec, t.address())
-        logg.info('add token transfer {} value {} from {} to {}'.format(
-            ts.symbol(),
-            token_value,
-            token_sender,
-            token_recipient,
-            )
-            )
-
-        db.add_transaction(
-                session,
-                r.transactionHash.hex(),
-                r.blockNumber,
-                r.transactionIndex,
-                w3.toChecksumAddress(token_sender),
-                w3.toChecksumAddress(token_recipient),
-                t.address(),
-                t.address(),
-                token_value,
-                token_value,
-                r.status == 1,
-                b.timestamp,
-                )
-        session.flush()
-
-
-    # TODO: simplify/ split up and/or comment, function is too long
-    def __process_convert(self, w3, session, t, r, l, b):
-        logg.warning('conversions are deactivated')
-        return
-#        token_source = l.topics[2][-20:].hex()
-#        token_source = w3.toChecksumAddress(token_source)
-#        token_destination = l.topics[3][-20:].hex()
-#        token_destination = w3.toChecksumAddress(token_destination)
-#        data_noox = l.data[2:]
-#        d = data_noox[:64]
-#        token_from_value = int(d, 16)
-#        d = data_noox[64:128]
-#        token_to_value = int(d, 16)
-#        token_trader = '0x' + data_noox[192-40:]
-#
-#        #ts = ContractRegistry.get_address(token_source)
-#        ts = CICRegistry.get_address(CICRegistry.bancor_chain_spec, t.address())
-#        #if ts == None:
-#        #    ts = ContractRegistry.reserves[token_source]
-#        td = ContractRegistry.get_address(token_destination)
-#        #if td == None:
-#        #    td = ContractRegistry.reserves[token_source]
-#        logg.info('add token convert {} -> {} value {} -> {} trader {}'.format(
-#            ts.symbol(),
-#            td.symbol(),
-#            token_from_value,
-#            token_to_value,
-#            token_trader,
-#            )
-#            )
-#
-#        db.add_transaction(
-#                session,
-#                r.transactionHash.hex(),
-#                r.blockNumber,
-#                r.transactionIndex,
-#                w3.toChecksumAddress(token_trader),
-#                w3.toChecksumAddress(token_trader),
-#                token_source,
-#                token_destination,
-#                r.status == 1,
-#                b.timestamp,
-#                )
-#        session.flush()
-
-
-    def check_token(self, address):
-            t = None
-            try:
-                t = CICRegistry.get_address(CICRegistry.default_chain_spec, address)
-                return t
-            except UnknownContractError:
-                logg.debug('contract {} not in registry'.format(address))
-
-            # If nothing was returned, we look up the token in the declarator
-            for trust in self.trusts:
-                logg.debug('look up declaration for contract {} with trust {}'.format(address, trust))
-                fn = self.declarator.function('declaration')
-                # TODO: cache trust in LRUcache
-                declaration_array = fn(trust, address).call()
-                try:
-                    declaration = to_token_declaration(trust, address, declaration_array, [rubberstamp])
-                    logg.debug('found declaration for token {} from trust address {}'.format(address, trust))
-                except UnknownDeclarationError:
-                    continue
-               
-                try:
-                    c = w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=address)
-                    t = CICRegistry.add_token(self.chain_spec, c)
-                    break
-                except ValueError:
-                    logg.error('declaration for {} validates as token, but location is not ERC20 compatible'.format(address))
-
-            return t
-
-
-    # TODO use input data instead of logs
-    def process(self, w3, session, block):
-        #self.refresh_registry(w3)
-        tx_count = w3.eth.getBlockTransactionCount(block.hash)
-        b = w3.eth.getBlock(block.hash)
-        for i in range(self.tx_height, tx_count):
-            tx = w3.eth.getTransactionByBlock(block.hash, i)
-            if tx.to == None:
-                logg.debug('block {} tx {} is contract creation tx, skipping'.format(block.number, i))
-                continue
-            if len(w3.eth.getCode(tx.to)) == 0:
-                logg.debug('block {} tx {} not a contract tx, skipping'.format(block.number, i))
-                continue
-
-            t = self.check_token(tx.to)
-            if t != None and isinstance(t, Token):
-                r = w3.eth.getTransactionReceipt(tx.hash)
-                for l in r.logs:
-                    logg.debug('block {} tx {} {} token log {} {}'.format(block.number, i, tx.hash.hex(), l.logIndex, l.topics[0].hex()))
-                    if l.topics[0].hex() == log_topics['transfer']:
-                        self.__process_tx(w3, session, t, r, l, b)
-
-            # TODO: cache contracts in LRUcache
-            elif self.convert_enabled and tx.to == CICRegistry.get_contract(CICRegistry.default_chain_spec, 'Converter').address:
-                r = w3.eth.getTransactionReceipt(tx.hash)
-                for l in r.logs:
-                    logg.info('block {} tx {} {} bancornetwork log {} {}'.format(block.number, i, tx.hash.hex(), l.logIndex, l.topics[0].hex()))
-                    if l.topics[0].hex() == log_topics['convert']:
-                        self.__process_convert(w3, session, t, r, l, b)
-            
-            session.execute("UPDATE tx_sync SET tx = '{}'".format(tx.hash.hex()))
-            session.commit()
-            self.tx_height += 1
-
-
-    def __get_next_retry(self, backoff=False):
-        return 1
-
-
-    def loop(self):
-        logg.info('starting at block {} tx index {}'.format(self.block_height, self.tx_height))
-        self.state = RunStateEnum.RUN
-        while self.state == RunStateEnum.RUN:
-            (provider, w3) = web3_constructor()
-            session = SessionBase.create_session()
-            try:
-                block = w3.eth.getBlock(self.block_height)
-                self.process(w3, session, block)
-                self.block_height += 1
-                self.tx_height = 0
-            except BlockNotFound as e:
-                logg.debug('no block {} yet, zZzZ...'.format(self.block_height))
-                time.sleep(self.__get_next_retry())
-            except ConnectionClosedError as e:
-                logg.info('connection gone, retrying')
-                time.sleep(self.__get_next_retry(True))
-            except OSError as e:
-                logg.error('cannot connect {}'.format(e))
-                time.sleep(self.__get_next_retry(True))
-            except Exception as e:
-                session.close()
-                raise(e)
-            session.close()
-
-
-    def load(self, w3):
-        session = SessionBase.create_session()
-        r = session.execute('SELECT tx FROM tx_sync').first()
-        if r != None:
-            if r[0] == '0x{0:0{1}X}'.format(0, 64):
-                logg.debug('last tx was zero-address, starting from scratch')
-                return
-            t = w3.eth.getTransaction(r[0])
-            
-            self.block_height = t.blockNumber
-            self.tx_height = t.transactionIndex+1
-            c = w3.eth.getBlockTransactionCount(t.blockHash.hex())
-            logg.debug('last tx processed {} index {} (max index {})'.format(t.blockNumber, t.transactionIndex, c-1))
-            if c == self.tx_height:
-                self.block_height += 1
-                self.tx_height = 0
-        session.close()
-
-(provider, w3) = web3_constructor()
-trust = config.get('CIC_TRUST_ADDRESS', "").split(",")
-chain_spec = args.i
-
-try:
-    w3.eth.chainId
-except Exception as e:
-    logg.exception(e)
-    sys.stderr.write('cannot connect to evm node\n')
-    sys.exit(1)
-
-def main():
-    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)
-
-    t = Tracker(chain_spec, trust) 
-    t.load(w3)
-    t.loop()
-    
-
-if __name__ == '__main__':
-    main()

apps/cic-cache/cic_cache/version.py

@@ -5,7 +5,7 @@ version = (
         0,
         2,
         0,
-        'alpha.1',
+        'alpha.2',
         )
 
 version_object = semver.VersionInfo(

apps/cic-cache/config/docker/cic.ini

@@ -0,0 +1,4 @@
+[cic]
+chain_spec =
+registry_address =  
+trust_address = 0xEb3907eCad74a0013c259D5874AE7f22DcBcC95C

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

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

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

@@ -1,3 +1,2 @@
 [eth]
 provider = ws://localhost:63546
-chain_id = 8996

apps/cic-cache/config/docker/syncer.ini

@@ -0,0 +1,2 @@
+[syncer]
+loop_interval = 1

apps/cic-cache/config/eth.ini

@@ -1,7 +1,2 @@
 [eth]
 provider = ws://localhost:8545
-#ttp_provider = http://localhost:8545
-#provider = http://localhost:8545
-gas_provider_address = 
-#chain_id = 
-abi_dir = /usr/local/share/cic/solidity/abi

apps/cic-cache/config/syncer.ini

@@ -0,0 +1,2 @@
+[syncer]
+loop_interval = 5

apps/cic-cache/docker/Dockerfile

@@ -17,7 +17,7 @@ RUN apt-get update && \
 
 # Copy shared requirements from top of mono-repo
 RUN echo "copying root req file ${root_requirement_file}"
-RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a44
+RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a53
 
 COPY cic-cache/requirements.txt ./
 COPY cic-cache/setup.cfg \

apps/cic-cache/requirements.txt

@@ -1,10 +1,13 @@
+cic-base~=0.1.2a53
 alembic==1.4.2
-confini~=0.3.6b2
+confini~=0.3.6rc3
 uwsgi==2.0.19.1
 moolb~=0.1.0
-cic-registry~=0.5.3a4
+cic-eth-registry~=0.5.4a9
 SQLAlchemy==1.3.20
 semver==2.13.0
 psycopg2==2.8.6
 celery==4.4.7
 redis==3.5.3
+chainlib~=0.0.1a44
+chainsyncer~=0.0.1a20

apps/cic-cache/setup.cfg

@@ -29,11 +29,13 @@ packages =
 	cic_cache.db
 	cic_cache.db.models
 	cic_cache.runnable
+	cic_cache.runnable.daemons
+	cic_cache.runnable.daemons.filters
 scripts =
 	./scripts/migrate.py
 
 [options.entry_points]
 console_scripts =
-	cic-cache-trackerd = cic_cache.runnable.tracker:main
-	cic-cache-serverd = cic_cache.runnable.server:main
-	cic-cache-taskerd = cic_cache.runnable.tasker:main
+	cic-cache-trackerd = cic_cache.runnable.daemons.tracker:main
+	cic-cache-serverd = cic_cache.runnable.daemons.server:main
+	cic-cache-taskerd = cic_cache.runnable.daemons.tasker:main

apps/cic-eth/cic_eth/api/api_admin.py

@@ -31,7 +31,10 @@ from cic_eth.db.models.tx import TxCache
 from cic_eth.db.models.nonce import Nonce
 from cic_eth.db.enum import (
         StatusEnum,
+        StatusBits,
         is_alive,
+        is_error_status,
+        status_str,
     )
 from cic_eth.error import InitializationError
 from cic_eth.db.error import TxStateChangeError
@@ -42,6 +45,8 @@ app = celery.current_app
 #logg = logging.getLogger(__file__)
 logg = logging.getLogger()
 
+local_fail = StatusBits.LOCAL_ERROR | StatusBits.NODE_ERROR | StatusBits.UNKNOWN_ERROR
+
 
 class AdminApi:
     """Provides an interface to view and manipulate existing transaction tasks and system runtime settings.
@@ -129,7 +134,8 @@ class AdminApi:
         return s_have.apply_async()
 
 
-    def resend(self, tx_hash_hex, chain_str, in_place=True, unlock=False):
+    def resend(self, tx_hash_hex, chain_spec, in_place=True, unlock=False):
+
         logg.debug('resend {}'.format(tx_hash_hex))
         s_get_tx_cache = celery.signature(
             'cic_eth.queue.tx.get_tx_cache',
@@ -151,7 +157,7 @@ class AdminApi:
         s = celery.signature(
             'cic_eth.eth.tx.resend_with_higher_gas',
             [
-                chain_str,
+                chain_spec.asdict(),
                 None,
                 1.01,
                 ],
@@ -171,7 +177,7 @@ class AdminApi:
             s_gas = celery.signature(
                 'cic_eth.admin.ctrl.unlock_send',
                 [
-                    chain_str,
+                    chain_spec.asdict(),
                     tx_dict['sender'],
                 ],
                 queue=self.queue,
@@ -195,6 +201,7 @@ class AdminApi:
         blocking_tx = None
         blocking_nonce = None
         nonce_otx = 0
+        last_nonce = -1
         for k in txs.keys():
             s_get_tx = celery.signature(
                     'cic_eth.queue.tx.get_tx',
@@ -205,18 +212,25 @@ class AdminApi:
                     )
             tx = s_get_tx.apply_async().get()
             #tx = get_tx(k)
-            logg.debug('checking nonce {}'.format(tx['nonce']))
-            if tx['status'] in [StatusEnum.REJECTED, StatusEnum.FUBAR]:
-                blocking_tx = k
-                blocking_nonce = tx['nonce']
+            logg.debug('checking nonce {} (previous {})'.format(tx['nonce'], last_nonce))
             nonce_otx = tx['nonce']
+            if not is_alive(tx['status']) and tx['status'] & local_fail > 0:
+                logg.info('permanently errored {} nonce {} status {}'.format(k, nonce_otx, status_str(tx['status'])))
+                blocking_tx = k
+                blocking_nonce = nonce_otx
+            elif nonce_otx - last_nonce > 1:
+                logg.error('nonce gap; {} followed {} for account {}'.format(nonce_otx, last_nonce, tx['from']))
+                blocking_tx = k
+                blocking_nonce = nonce_otx
+                break
+            last_nonce = nonce_otx
 
-        nonce_cache = Nonce.get(address)
+        #nonce_cache = Nonce.get(address)
         #nonce_w3 = self.w3.eth.getTransactionCount(address, 'pending') 
         
         return {
             'nonce': {
-                'network': nonce_cache,
+                #'network': nonce_cache,
                 'queue': nonce_otx,
                 #'cache': nonce_cache,
                 'blocking': blocking_nonce,
@@ -270,16 +284,15 @@ class AdminApi:
 #        self.w3.eth.sign(addr, text='666f6f')
 
 
-    def account(self, chain_spec, address, cols=['tx_hash', 'sender', 'recipient', 'nonce', 'block', 'tx_index', 'status', 'network_status', 'date_created'], include_sender=True, include_recipient=True):
+    def account(self, chain_spec, address, include_sender=True, include_recipient=True, renderer=None, w=sys.stdout):
         """Lists locally originated transactions for the given Ethereum address.
 
         Performs a synchronous call to the Celery task responsible for performing the query.
 
         :param address: Ethereum address to return transactions for
         :type address: str, 0x-hex
-        :param cols: Data columns to include
-        :type cols: list of str
         """
+        last_nonce = -1
         s = celery.signature(
                 'cic_eth.queue.tx.get_account_tx',
                 [
@@ -291,33 +304,45 @@ class AdminApi:
 
         tx_dict_list = []
         for tx_hash in txs.keys():
+            errors = []
             s = celery.signature(
                     'cic_eth.queue.tx.get_tx_cache',
                     [tx_hash],
                     queue=self.queue,
                     )
             tx_dict = s.apply_async().get()
-            if tx_dict['sender'] == address and not include_sender:
-                logg.debug('skipping sender tx {}'.format(tx_dict['tx_hash']))
-                continue
+            if tx_dict['sender'] == address:
+                if tx_dict['nonce'] - last_nonce > 1:
+                    logg.error('nonce gap; {} followed {} for address {} tx {}'.format(tx_dict['nonce'], last_nonce, tx_dict['sender'], tx_hash))
+                    errors.append('nonce')
+                elif tx_dict['nonce'] == last_nonce:
+                    logg.info('nonce {} duplicate for address {} in tx {}'.format(tx_dict['nonce'], tx_dict['sender'], tx_hash))
+                last_nonce = tx_dict['nonce']
+                if not include_sender:
+                    logg.debug('skipping sender tx {}'.format(tx_dict['tx_hash']))
+                    continue
             elif tx_dict['recipient'] == address and not include_recipient:
                 logg.debug('skipping recipient tx {}'.format(tx_dict['tx_hash']))
                 continue
 
-            logg.debug(tx_dict)
             o = {
                 'nonce': tx_dict['nonce'], 
                 'tx_hash': tx_dict['tx_hash'],
                 'status': tx_dict['status'],
                 'date_updated': tx_dict['date_updated'],
+                'errors': errors,
                     }
-            tx_dict_list.append(o)
+            if renderer != None:
+                r = renderer(o)
+                w.write(r + '\n')
+            else:
+                tx_dict_list.append(o)
 
         return tx_dict_list
 
 
     # TODO: Add exception upon non-existent tx aswell as invalid tx data to docstring 
-    def tx(self, chain_spec, tx_hash=None, tx_raw=None, registry=None):
+    def tx(self, chain_spec, tx_hash=None, tx_raw=None, registry=None, renderer=None, w=sys.stdout):
         """Output local and network details about a given transaction with local origin.
 
         If the transaction hash is given, the raw trasnaction data will be retrieved from the local transaction queue backend. Otherwise the raw transaction data must be provided directly. Only one of transaction hash and transaction data can be passed.
@@ -456,15 +481,17 @@ class AdminApi:
             tx['destination_token_symbol'] = destination_token.symbol()
             tx['recipient_token_balance'] = source_token.function('balanceOf')(tx['recipient']).call()
 
-        tx['network_status'] = 'Not submitted'
+        # TODO: this can mean either not subitted or culled, need to check other txs with same nonce to determine which
+        tx['network_status'] = 'Not in node' 
 
         r = None
         try:
             o = transaction(tx_hash)
             r = self.rpc.do(o)
+            if r != None:
+                tx['network_status'] = 'Mempool'
         except Exception as e:
             logg.warning('(too permissive exception handler, please fix!) {}'.format(e))
-            tx['network_status'] = 'Mempool'
 
         if r != None:
             try:
@@ -511,4 +538,9 @@ class AdminApi:
             for p in problems:
                 sys.stderr.write('!!!{}\n'.format(p))
 
-        return tx
+        if renderer == None:
+            return tx
+
+        r = renderer(tx)
+        w.write(r + '\n')
+        return None

apps/cic-eth/cic_eth/eth/tx.py

@@ -7,7 +7,10 @@ import requests
 from chainlib.eth.constant import ZERO_ADDRESS
 from chainlib.chain import ChainSpec
 from chainlib.eth.address import is_checksum_address
-from chainlib.eth.gas import balance
+from chainlib.eth.gas import (
+        balance,
+        price,
+        )
 from chainlib.eth.error import (
         EthException,
         NotFoundEthException,
@@ -17,11 +20,15 @@ from chainlib.eth.tx import (
         receipt,
         raw,
         TxFormat,
+        TxFactory,
         unpack,
         )
 from chainlib.connection import RPCConnection
 from chainlib.hash import keccak256_hex_to_hex
-from chainlib.eth.gas import Gas
+from chainlib.eth.gas import (
+        Gas,
+        OverrideGasOracle,
+        )
 from chainlib.eth.contract import (
         abi_decode_single,
         ABIContractType,
@@ -52,6 +59,7 @@ from cic_eth.queue.tx import (
         get_tx,
         register_tx,
         get_nonce_tx,
+        create as queue_create,
         )
 from cic_eth.error import OutOfGasError
 from cic_eth.error import LockedError
@@ -281,6 +289,7 @@ def send(self, txs, chain_spec_dict):
     o = raw(tx_hex)
     conn = RPCConnection.connect(chain_spec, 'default')
     conn.do(o)
+
     s_set_sent.apply_async()
 
     tx_tail = txs[1:]
@@ -369,7 +378,7 @@ def refill_gas(self, recipient_address, chain_spec_dict):
 
 
 @celery_app.task(bind=True, base=CriticalSQLAlchemyAndSignerTask)
-def resend_with_higher_gas(self, txold_hash_hex, chain_str, gas=None, default_factor=1.1):
+def resend_with_higher_gas(self, txold_hash_hex, chain_spec_dict, gas=None, default_factor=1.1):
     """Create a new transaction from an existing one with same nonce and higher gas price.
 
     :param txold_hash_hex: Transaction to re-create
@@ -393,46 +402,55 @@ def resend_with_higher_gas(self, txold_hash_hex, chain_str, gas=None, default_fa
         session.close()
         raise NotLocalTxError(txold_hash_hex)
 
-    chain_spec = ChainSpec.from_chain_str(chain_str)
-    c = RpcClient(chain_spec)
+    chain_spec = ChainSpec.from_dict(chain_spec_dict)
 
     tx_signed_raw_bytes = bytes.fromhex(otx.signed_tx[2:])
     tx = unpack(tx_signed_raw_bytes, chain_spec.chain_id())
     logg.debug('resend otx {} {}'.format(tx, otx.signed_tx))
 
-    queue = self.request.delivery_info['routing_key']
+    queue = self.request.delivery_info.get('routing_key')
 
     logg.debug('before {}'.format(tx))
-    if gas != None:
-        tx['gasPrice'] = gas
-    else:
-        gas_price = c.gas_price()
-        if tx['gasPrice'] > gas_price:
-            logg.info('Network gas price {} is lower than overdue tx gas price {}'.format(gas_price, tx['gasPrice']))
+
+    rpc = RPCConnection.connect(chain_spec, 'default')
+    new_gas_price = gas
+    if new_gas_price == None:
+        o = price()
+        r = rpc.do(o)
+        current_gas_price = int(r, 16)
+        if tx['gasPrice'] > current_gas_price:
+            logg.info('Network gas price {} is lower than overdue tx gas price {}'.format(curent_gas_price, tx['gasPrice']))
             #tx['gasPrice'] = int(tx['gasPrice'] * default_factor)
-            tx['gasPrice'] += 1
+            new_gas_price = tx['gasPrice'] + 1
         else:
             new_gas_price = int(tx['gasPrice'] * default_factor)
-            if gas_price > new_gas_price:
-                tx['gasPrice'] = gas_price
-            else:
-                tx['gasPrice'] = new_gas_price
+            #if gas_price > new_gas_price:
+            #    tx['gasPrice'] = gas_price
+            #else:
+            #    tx['gasPrice'] = new_gas_price
 
-    (tx_hash_hex, tx_signed_raw_hex) = sign_tx(tx, chain_str)
+
+    rpc_signer = RPCConnection.connect(chain_spec, 'signer')
+    gas_oracle = OverrideGasOracle(price=new_gas_price, conn=rpc)
+
+    c = TxFactory(signer=rpc_signer, gas_oracle=gas_oracle, chain_id=chain_spec.chain_id())
+    logg.debug('change gas price from old {} to new {} for tx {}'.format(tx['gasPrice'], new_gas_price, tx))
+    tx['gasPrice'] = new_gas_price
+    (tx_hash_hex, tx_signed_raw_hex) = c.build_raw(tx)
     queue_create(
         tx['nonce'],
         tx['from'],
         tx_hash_hex,
         tx_signed_raw_hex,
-        chain_str,
+        chain_spec,
         session=session,
             )
     TxCache.clone(txold_hash_hex, tx_hash_hex, session=session)
     session.close()
 
-    s = create_check_gas_and_send_task(
+    s = create_check_gas_task(
             [tx_signed_raw_hex],
-            chain_str, 
+            chain_spec, 
             tx['from'],
             tx['gasPrice'] * tx['gas'],
             [tx_hash_hex],

apps/cic-eth/cic_eth/queue/tx.py

@@ -676,6 +676,7 @@ def get_status_tx(status, not_status=None, before=None, exact=False, limit=0, se
         q = q.filter(Otx.status.op('&')(status)>0)
         if not_status != None:
             q = q.filter(Otx.status.op('&')(not_status)==0)
+    q = q.order_by(Otx.nonce.asc(), Otx.date_created.asc())
     i = 0
     for o in q.all():
         if limit > 0 and i == limit:
@@ -687,7 +688,7 @@ def get_status_tx(status, not_status=None, before=None, exact=False, limit=0, se
 
 
 # TODO: move query to model
-def get_upcoming_tx(status=StatusEnum.READYSEND, recipient=None, before=None, chain_id=0, session=None):
+def get_upcoming_tx(status=StatusEnum.READYSEND, not_status=None, recipient=None, before=None, limit=0, chain_id=0, session=None):
     """Returns the next pending transaction, specifically the transaction with the lowest nonce, for every recipient that has pending transactions.
 
     Will omit addresses that have the LockEnum.SEND bit in Lock set.
@@ -721,7 +722,10 @@ def get_upcoming_tx(status=StatusEnum.READYSEND, recipient=None, before=None, ch
     if status == StatusEnum.PENDING:
         q_outer = q_outer.filter(Otx.status==status.value)
     else:
-        q_outer = q_outer.filter(Otx.status.op('&')(status.value)==status.value)
+        q_outer = q_outer.filter(Otx.status.op('&')(status)==status)
+
+    if not_status != None:
+        q_outer = q_outer.filter(Otx.status.op('&')(not_status)==0)
 
     if recipient != None:
         q_outer = q_outer.filter(TxCache.recipient==recipient)
@@ -730,6 +734,7 @@ def get_upcoming_tx(status=StatusEnum.READYSEND, recipient=None, before=None, ch
 
     txs = {}
 
+    i = 0
     for r in q_outer.all():
         q = session.query(Otx)
         q = q.join(TxCache)
@@ -758,6 +763,10 @@ def get_upcoming_tx(status=StatusEnum.READYSEND, recipient=None, before=None, ch
         session.add(o)
         session.commit()
 
+        i += 1
+        if limit > 0 and limit == i:
+            break
+
     SessionBase.release_session(session)
 
     return txs

apps/cic-eth/cic_eth/registry.py

@@ -0,0 +1,33 @@
+# standard imports
+import logging
+
+# external imports
+from cic_eth_registry import CICRegistry
+from cic_eth_registry.lookup.declarator import AddressDeclaratorLookup
+from cic_eth_registry.lookup.tokenindex import TokenIndexLookup
+
+logg = logging.getLogger()
+
+
+def connect_token_registry(rpc, chain_spec):
+    registry = CICRegistry(chain_spec, rpc)
+    token_registry_address = registry.by_name('TokenRegistry')
+    logg.debug('using token registry address {}'.format(token_registry_address))
+    lookup = TokenIndexLookup(token_registry_address)
+    CICRegistry.add_lookup(lookup)
+
+
+def connect_declarator(rpc, chain_spec, trusted_addresses):
+    registry = CICRegistry(chain_spec, rpc)
+    declarator_address = registry.by_name('AddressDeclarator')
+    logg.debug('using declarator address {}'.format(declarator_address))
+    lookup = AddressDeclaratorLookup(declarator_address, trusted_addresses)
+    CICRegistry.add_lookup(lookup)
+
+
+def connect(rpc, chain_spec, registry_address):
+    CICRegistry.address = registry_address
+    registry = CICRegistry(chain_spec, rpc)
+    registry_address = registry.by_name('ContractRegistry')
+    
+    return registry

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

@@ -88,7 +88,7 @@ run = True
 
 class DispatchSyncer:
 
-    yield_delay = 0.005
+    yield_delay = 0.0005
 
     def __init__(self, chain_spec):
         self.chain_spec = chain_spec

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

@@ -1,3 +1,4 @@
+# standard imports
 import os
 import sys
 import logging
@@ -5,22 +6,36 @@ import argparse
 import re
 import datetime
 
-import web3
+# external imports
 import confini
 import celery
 from cic_eth_registry import CICRegistry
 from chainlib.chain import ChainSpec
+from chainlib.eth.tx import unpack
+from chainlib.connection import RPCConnection
+from chainlib.eth.block import (
+        block_latest,
+        block_by_number,
+        Block,
+        )
+from chainsyncer.driver import HeadSyncer
+from chainsyncer.backend import MemBackend
+from chainsyncer.error import NoBlockForYou
 
+# local imports
 from cic_eth.db import dsn_from_config
 from cic_eth.db import SessionBase
-from cic_eth.eth import RpcClient
-from cic_eth.sync.retry import RetrySyncer
-from cic_eth.queue.tx import get_status_tx
-from cic_eth.queue.tx import get_tx
+from cic_eth.queue.tx import (
+        get_status_tx,
+        get_tx,
+#        get_upcoming_tx,
+        )
 from cic_eth.admin.ctrl import lock_send
-from cic_eth.db.enum import StatusEnum
-from cic_eth.db.enum import LockEnum
-from cic_eth.eth.util import unpack_signed_raw_tx_hex
+from cic_eth.db.enum import (
+        StatusEnum,
+        StatusBits,
+        LockEnum,
+        )
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -31,7 +46,8 @@ argparser = argparse.ArgumentParser(description='daemon that monitors transactio
 argparser.add_argument('-p', '--provider', dest='p', type=str, help='rpc provider')
 argparser.add_argument('-c', type=str, default=config_dir, help='config root to use')
 argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='chain spec')
-argparser.add_argument('--retry-delay', dest='retry_delay', type=str, help='seconds to wait for retrying a transaction that is marked as sent')
+argparser.add_argument('--batch-size', dest='batch_size', type=int, default=50, help='max amount of txs to resend per iteration')
+argparser.add_argument('--retry-delay', dest='retry_delay', type=int, help='seconds to wait for retrying a transaction that is marked as sent')
 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')
@@ -51,7 +67,6 @@ config.process()
 # override args
 args_override = {
         'ETH_PROVIDER': getattr(args, 'p'),
-        'ETH_ABI_DIR': getattr(args, 'abi_dir'),
         'CIC_CHAIN_SPEC': getattr(args, 'i'),
         'CIC_TX_RETRY_DELAY': getattr(args, 'retry_delay'),
         }
@@ -59,6 +74,7 @@ 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.batch_size, '_BATCH_SIZE', True)
 
 app = celery.Celery(backend=config.get('CELERY_RESULT_URL'),  broker=config.get('CELERY_BROKER_URL'))
 
@@ -66,10 +82,10 @@ queue = args.q
 
 chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
 
-RPCConnection.registry_location(args.p, chain_spec, tag='default')
+RPCConnection.register_location(config.get('ETH_PROVIDER'), chain_spec, tag='default')
 
 dsn = dsn_from_config(config)
-SessionBase.connect(dsn)
+SessionBase.connect(dsn, debug=config.true('DATABASE_DEBUG'))
 
 straggler_delay = int(config.get('CIC_TX_RETRY_DELAY'))
 
@@ -178,53 +194,85 @@ def dispatch(conn, chain_spec):
 #    s_send.apply_async()
 
 
-class RetrySyncer(Syncer):
+class StragglerFilter:
 
-    def __init__(self, chain_spec, stalled_grace_seconds, failed_grace_seconds=None, final_func=None):
+    def __init__(self, chain_spec, queue='cic-eth'):
+        self.chain_spec = chain_spec
+        self.queue = queue
+
+
+    def filter(self, conn, block, tx, db_session=None):
+        logg.debug('tx {}'.format(tx))
+        s_send = celery.signature(
+                'cic_eth.eth.tx.resend_with_higher_gas',
+                [
+                    tx,
+                    self.chain_spec.asdict(),
+                ],
+                queue=self.queue,
+        )
+        return s_send.apply_async()
+        #return s_send
+
+
+    def __str__(self):
+        return 'stragglerfilter'
+
+
+class RetrySyncer(HeadSyncer):
+
+    def __init__(self, conn, chain_spec, stalled_grace_seconds, batch_size=50, failed_grace_seconds=None):
+        backend = MemBackend(chain_spec, None)
+        super(RetrySyncer, self).__init__(backend)
         self.chain_spec = chain_spec
         if failed_grace_seconds == None:
             failed_grace_seconds = stalled_grace_seconds
         self.stalled_grace_seconds = stalled_grace_seconds
         self.failed_grace_seconds = failed_grace_seconds
-        self.final_func = final_func
+        self.batch_size = batch_size
+        self.conn = conn
 
 
-    def get(self):
-#            before = datetime.datetime.utcnow() - datetime.timedelta(seconds=self.failed_grace_seconds)
-#            failed_txs = get_status_tx(
-#                    StatusEnum.SENDFAIL.value,
-#                    before=before,
-#                    )
-            before = datetime.datetime.utcnow() - datetime.timedelta(seconds=self.stalled_grace_seconds)
-            stalled_txs = get_status_tx(
-                    StatusBits.IN_NETWORK.value,
-                    not_status=StatusBits.FINAL | StatusBits.MANUAL | StatusBits.OBSOLETE,
-                    before=before,
-                    )
-       #     return list(failed_txs.keys()) + list(stalled_txs.keys())
-            return stalled_txs
-
-    def process(self, conn, ref):
-        logg.debug('tx {}'.format(ref))
-        for f in self.filter:
-            f(conn, ref, None, str(self.chain_spec))
+    def get(self, conn):
+        o = block_latest()
+        r = conn.do(o)
+        (pair, flags) = self.backend.get()
+        n = int(r, 16)
+        if n == pair[0]:
+            raise NoBlockForYou('block {} already checked'.format(n))
+        o = block_by_number(n)
+        r = conn.do(o)
+        b = Block(r)
+        return b
 
 
+    def process(self, conn, block):
+        before = datetime.datetime.utcnow() - datetime.timedelta(seconds=self.stalled_grace_seconds)
+        stalled_txs = get_status_tx(
+                StatusBits.IN_NETWORK.value,
+                not_status=StatusBits.FINAL | StatusBits.MANUAL | StatusBits.OBSOLETE,
+                before=before,
+                limit=self.batch_size,
+                )
+#        stalled_txs = get_upcoming_tx(
+#                status=StatusBits.IN_NETWORK.value, 
+#                not_status=StatusBits.FINAL | StatusBits.MANUAL | StatusBits.OBSOLETE,
+#                before=before,
+#                limit=self.batch_size,
+#                )
+        for tx in stalled_txs:
+            self.filter.apply(self.conn, block, tx)
+        self.backend.set(block.number, 0)
 
-    def loop(self, interval):
-        while self.running and Syncer.running_global:
-            rpc = RPCConnection.connect(self.chain_spec, 'default')
-            for tx in self.get():
-                self.process(rpc, tx)
-            if self.final_func != None:
-                self.final_func(rpc, self.chain_spec)
-            time.sleep(interval)
 
 def main(): 
-
-    syncer = RetrySyncer(chain_spec, straggler_delay, final_func=dispatch)
-    syncer.filter.append(sendfail_filter)
-    syncer.loop(float(straggler_delay))
+    #o = block_latest()
+    conn = RPCConnection.connect(chain_spec, 'default')
+    #block = conn.do(o)
+    syncer = RetrySyncer(conn, chain_spec, straggler_delay, batch_size=config.get('_BATCH_SIZE'))
+    syncer.backend.set(0, 0)
+    syncer.add_filter(StragglerFilter(chain_spec, queue=queue))
+    syncer.loop(float(straggler_delay), conn)
 
 
 if __name__ == '__main__':

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

@@ -16,8 +16,6 @@ from chainlib.eth.connection import EthUnixSignerConnection
 from chainlib.chain import ChainSpec
 
 # local imports
-from cic_eth_registry import CICRegistry
-
 from cic_eth.eth import erc20
 from cic_eth.eth import tx
 from cic_eth.eth import account
@@ -33,6 +31,11 @@ from cic_eth.db.models.base import SessionBase
 from cic_eth.db.models.otx import Otx
 from cic_eth.db import dsn_from_config
 from cic_eth.ext import tx
+from cic_eth.registry import (
+        connect as connect_registry,
+        connect_declarator,
+        connect_token_registry,
+        )
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -121,8 +124,6 @@ RPCConnection.register_location(config.get('SIGNER_SOCKET_PATH'), chain_spec, 's
 
 Otx.tracing = config.true('TASKS_TRACE_QUEUE_STATUS')
 
-CICRegistry.address = config.get('CIC_REGISTRY_ADDRESS')
-
 
 def main():
     argv = ['worker']
@@ -145,8 +146,8 @@ def main():
 #        Callback.ssl_ca_file = config.get('SSL_CA_FILE')
 
     rpc = RPCConnection.connect(chain_spec, 'default')
-    registry = CICRegistry(chain_spec, rpc)
-    registry_address = registry.by_name('ContractRegistry')
+
+    connect_registry(rpc, chain_spec, config.get('CIC_REGISTRY_ADDRESS'))
 
     trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
     if trusted_addresses_src == None:
@@ -155,6 +156,8 @@ def main():
     trusted_addresses = trusted_addresses_src.split(',')
     for address in trusted_addresses:
         logg.info('using trusted address {}'.format(address))
+    connect_declarator(rpc, chain_spec, trusted_addresses)
+    connect_token_registry(rpc, chain_spec)
     
     current_app.worker_main(argv)
 

apps/cic-eth/cic_eth/runnable/resend.py

@@ -7,13 +7,10 @@ import os
 # third-party imports
 import celery
 import confini
-import web3
-from cic_registry import CICRegistry
-from cic_registry.chain import ChainSpec
-from cic_registry.chain import ChainRegistry
+from chainlib.chain import ChainSpec
+from chainlib.eth.connection import EthHTTPConnection
 
 # local imports
-from cic_eth.eth.rpc import RpcClient
 from cic_eth.api.api_admin import AdminApi
 
 logging.basicConfig(level=logging.WARNING)
@@ -55,41 +52,20 @@ args_override = {
 config.censor('PASSWORD', 'DATABASE')
 config.censor('PASSWORD', 'SSL')
 logg.debug('config loaded from {}:\n{}'.format(config_dir, config))
+config.add(args.tx_hash, '_TX_HASH', True)
+config.add(args.unlock, '_UNLOCK', True)
 
 chain_spec = ChainSpec.from_chain_str(args.i)
-chain_str = str(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 = web3.Web3.WebsocketProvider(blockchain_provider)
-elif re.match(re_http, blockchain_provider) != None:
-    blockchain_provider = web3.Web3.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)
+rpc = EthHTTPConnection(config.get('ETH_PROVIDER'))
 
 celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'))
 
-c = RpcClient(chain_spec)
-
-CICRegistry.init(c.w3, config.get('CIC_REGISTRY_ADDRESS'), chain_spec)
-chain_registry = ChainRegistry(chain_spec)
-CICRegistry.add_chain_registry(chain_registry)
-CICRegistry.add_path(config.get('ETH_ABI_DIR'))
-CICRegistry.load_for(chain_spec)
-
-
 def main():
-    api = AdminApi(c)
+    api = AdminApi(rpc)
     tx_details = api.tx(chain_spec, args.tx_hash)
-    t = api.resend(args.tx_hash, chain_str, unlock=True)
-
+    t = api.resend(args.tx_hash, chain_spec, unlock=config.get('_UNLOCK'))
+    print(t.get_leaf())
 
 if __name__ == '__main__':
     main()

apps/cic-eth/cic_eth/runnable/transfer.py

@@ -0,0 +1,100 @@
+#!/usr/bin/python
+import sys
+import os
+import logging
+import uuid
+import json
+import argparse
+
+# external imports
+import celery
+import confini
+import redis
+from xdg.BaseDirectory import xdg_config_home
+from chainlib.eth.address import to_checksum_address
+
+# local imports
+from cic_eth.api import Api
+
+logging.basicConfig(level=logging.WARNING)
+logg = logging.getLogger('create_account_script')
+logging.getLogger('confini').setLevel(logging.WARNING)
+logging.getLogger('gnupg').setLevel(logging.WARNING)
+
+default_config_dir = os.environ.get('CONFINI_DIR', '/usr/local/etc/cic')
+
+argparser = argparse.ArgumentParser()
+argparser.add_argument('--no-register', dest='no_register', action='store_true', help='Do not register new account in on-chain accounts index')
+argparser.add_argument('-c', type=str, default=default_config_dir, help='config file')
+argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='chain spec')
+argparser.add_argument('--token-symbol', dest='token_symbol', type=str, help='Symbol of token to transfer')
+argparser.add_argument('--redis-host', dest='redis_host', type=str, help='redis host to use for task submission')
+argparser.add_argument('--redis-port', dest='redis_port', type=int, help='redis host to use for task submission')
+argparser.add_argument('--redis-db', dest='redis_db', type=int, help='redis db to use for task submission and callback')
+argparser.add_argument('--redis-host-callback', dest='redis_host_callback', default='localhost', type=str, help='redis host to use for callback')
+argparser.add_argument('--redis-port-callback', dest='redis_port_callback', default=6379, type=int, help='redis port to use for callback')
+argparser.add_argument('--timeout', default=20.0, type=float, help='Callback timeout')
+argparser.add_argument('-q', type=str, default='cic-eth', help='Task queue')
+argparser.add_argument('-v', action='store_true', help='Be verbose')
+argparser.add_argument('-vv', action='store_true', help='Be more verbose')
+argparser.add_argument('sender', type=str, help='Transaction sender')
+argparser.add_argument('recipient', type=str, help='Transaction recipient')
+argparser.add_argument('value', type=int, help='Transaction value with decimals')
+args = argparser.parse_args()
+
+if args.vv:
+    logg.setLevel(logging.DEBUG)
+if args.v:
+    logg.setLevel(logging.INFO)
+
+config_dir = args.c
+config = confini.Config(config_dir, os.environ.get('CONFINI_ENV_PREFIX'))
+config.process()
+args_override = {
+        'CIC_CHAIN_SPEC': getattr(args, 'i'),
+        'REDIS_HOST': getattr(args, 'redis_host'),
+        'REDIS_PORT': getattr(args, 'redis_port'),
+        'REDIS_DB': getattr(args, 'redis_db'),
+        }
+config.dict_override(args_override, 'cli')
+config.add(to_checksum_address(args.sender), '_SENDER', True)
+config.add(to_checksum_address(args.recipient), '_RECIPIENT', True)
+config.add(args.value, '_VALUE', True)
+config.add(args.token_symbol, '_SYMBOL', True)
+if config.get('_SYMBOL') == None:
+    raise ValueError('gas transfers not yet supported; token symbol required')
+celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'))
+
+
+def main():
+    redis_host = config.get('REDIS_HOST')
+    redis_port = config.get('REDIS_PORT')
+    redis_db = config.get('REDIS_DB')
+    redis_channel = str(uuid.uuid4())
+    r = redis.Redis(redis_host, redis_port, redis_db)
+
+    ps = r.pubsub()
+    ps.subscribe(redis_channel)
+    ps.get_message()
+
+    api = Api(
+            config.get('CIC_CHAIN_SPEC'),
+            queue=args.q,
+            callback_param='{}:{}:{}:{}'.format(args.redis_host_callback, args.redis_port_callback, redis_db, redis_channel),
+            callback_task='cic_eth.callbacks.redis.redis',
+            callback_queue=args.q,
+            )
+
+    #register = not args.no_register
+    #logg.debug('register {}'.format(register))
+    #t = api.create_account(register=register)
+    t = api.transfer(config.get('_SENDER'), config.get('_RECIPIENT'), config.get('_VALUE'), config.get('_SYMBOL'))
+
+    ps.get_message()
+    o = ps.get_message(timeout=args.timeout)
+    m = json.loads(o['data'])
+    print(m['result'])
+
+
+if __name__ == '__main__':
+    main()

apps/cic-eth/cic_eth/runnable/view.py

@@ -14,8 +14,6 @@ import datetime
 # external imports
 import confini
 import celery
-from cic_eth_registry import CICRegistry
-from cic_eth_registry.lookup.declarator import AddressDeclaratorLookup
 from chainlib.chain import ChainSpec
 from chainlib.eth.connection import EthHTTPConnection
 from hexathon import add_0x
@@ -27,6 +25,7 @@ from cic_eth.db.enum import (
     status_str,
     LockEnum,
 )
+from cic_eth.registry import connect as connect_registry
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -116,12 +115,16 @@ def render_tx(o, **kwargs):
     return content
 
 def render_account(o, **kwargs):
-    return '{} {} {} {}'.format(
+    s = '{} {} {} {}'.format(
             o['date_updated'],
             o['nonce'],
             o['tx_hash'],
             o['status'],
             )
+    if len(o['errors']) > 0:
+        s += ' !{}'.format(','.join(o['errors']))
+
+    return s
 
 
 def render_lock(o, **kwargs):
@@ -143,44 +146,31 @@ def render_lock(o, **kwargs):
     return s
 
 
-def connect_registry(registry_address, chain_spec, rpc):
-    CICRegistry.address = registry_address
-    registry = CICRegistry(chain_spec, rpc)
-    declarator_address = registry.by_name('AddressDeclarator')
-    lookup = AddressDeclaratorLookup(declarator_address, trusted_addresses)
-    registry.add_lookup(lookup)
-    return registry
-
-
 # TODO: move each command to submodule
 def main():
     txs  = []
     renderer = render_tx
     if len(config.get('_QUERY')) > 66:
-        registry = connect_registry(registry_address, chain_spec, rpc)
-        txs = [admin_api.tx(chain_spec, tx_raw=config.get('_QUERY'), registry=registry)]
+        registry = connect_registry(rpc, chain_spec, registry_address)
+        admin_api.tx(chain_spec, tx_raw=config.get('_QUERY'), registry=registry, renderer=renderer)
     elif len(config.get('_QUERY')) > 42:
-        registry = connect_registry(registry_address, chain_spec, rpc)
-        txs = [admin_api.tx(chain_spec, tx_hash=config.get('_QUERY'), registry=registry)]
+        registry = connect_registry(rpc, chain_spec, registry_address)
+        admin_api.tx(chain_spec, tx_hash=config.get('_QUERY'), registry=registry, renderer=renderer)
+
     elif len(config.get('_QUERY')) == 42:
-        registry = connect_registry(registry_address, chain_spec, rpc)
-        txs = admin_api.account(chain_spec, config.get('_QUERY'), include_recipient=False)
+        registry = connect_registry(rpc, chain_spec, registry_address)
+        txs = admin_api.account(chain_spec, config.get('_QUERY'), include_recipient=False, renderer=render_account)
         renderer = render_account
     elif len(config.get('_QUERY')) >= 4 and config.get('_QUERY')[:4] == 'lock':
         t = admin_api.get_lock()
         txs = t.get()
         renderer = render_lock
+        for tx in txs:
+            r = renderer(txs)
+            sys.stdout.write(r + '\n')
     else:
         raise ValueError('cannot parse argument {}'.format(config.get('_QUERY')))
+                   
 
-    if len(txs) == 0:
-        logg.info('no matches found')
-    else:
-        if fmt == 'json':
-            sys.stdout.write(json.dumps(txs))
-        else:
-            m = map(renderer, txs)
-            print(*m, sep="\n")
-                    
 if __name__ == '__main__':
     main()

apps/cic-eth/cic_eth/version.py

@@ -10,7 +10,7 @@ version = (
         0,
         10,
         1,
-        'beta.1',
+        'beta.2',
         )
 
 version_object = semver.VersionInfo(

apps/cic-eth/requirements.txt

@@ -1,14 +1,14 @@
-cic-base~=0.1.2a46
+cic-base~=0.1.2a53
 celery==4.4.7
 crypto-dev-signer~=0.4.14a16
 confini~=0.3.6rc3
-cic-eth-registry~=0.5.4a7
+cic-eth-registry~=0.5.4a9
 #cic-bancor~=0.0.6
 redis==3.5.3
 alembic==1.4.2
 websockets==8.1
 requests~=2.24.0
-eth_accounts_index~=0.0.11a3
+eth_accounts_index~=0.0.11a6
 erc20-transfer-authorization~=0.3.1a2
 #simple-rlp==0.1.2
 uWSGI==2.0.19.1
@@ -16,9 +16,9 @@ semver==2.13.0
 websocket-client==0.57.0
 moolb~=0.1.1b2
 eth-address-index~=0.1.1a5
-chainlib~=0.0.1a42
+chainlib~=0.0.1a43
 hexathon~=0.0.1a7
 chainsyncer~=0.0.1a20
 pysha3==1.0.2
 coincurve==15.0.0
-sarafu-faucet==0.0.2a13
+sarafu-faucet~=0.0.2a15

apps/cic-eth/setup.cfg

@@ -54,3 +54,4 @@ console_scripts =
 	# TODO: Merge this with ctl when subcmds sorted to submodules
 	cic-eth-tag = cic_eth.runnable.tag:main
 	cic-eth-resend = cic_eth.runnable.resend:main
+	cic-eth-transfer = cic_eth.runnable.transfer:main

apps/cic-eth/tests/task/test_task_tx.py

@@ -12,10 +12,12 @@ from chainlib.eth.tx import (
         transaction,
         receipt,
         )
+from hexathon import strip_0x
 
 # local imports
 from cic_eth.queue.tx import register_tx
 from cic_eth.eth.tx import cache_gas_data
+from cic_eth.db.models.otx import Otx
 
 logg = logging.getLogger()
 
@@ -60,6 +62,7 @@ def test_tx_send(
     assert rcpt['status'] == 1
 
 
+@pytest.mark.skip()
 def test_sync_tx(
         default_chain_spec,
         eth_rpc,
@@ -67,3 +70,44 @@ def test_sync_tx(
         celery_worker,
         ):
     pass
+
+
+def test_resend_with_higher_gas(
+        init_database,
+        default_chain_spec,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        celery_worker,
+        ):
+
+    chain_id = default_chain_spec.chain_id()
+    nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], eth_rpc)
+    c = Gas(signer=eth_signer, nonce_oracle=nonce_oracle, chain_id=chain_id)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 1024, tx_format=TxFormat.RLP_SIGNED)
+    #unpack(bytes.fromhex(strip_0x(tx_signed_raw_hex)), chain_id)
+    register_tx(tx_hash_hex, tx_signed_raw_hex, default_chain_spec, None, session=init_database)
+    cache_gas_data(tx_hash_hex, tx_signed_raw_hex, default_chain_spec.asdict())
+    tx_before = unpack(bytes.fromhex(strip_0x(tx_signed_raw_hex)), default_chain_spec.chain_id())
+
+    s = celery.signature(
+            'cic_eth.eth.tx.resend_with_higher_gas',
+            [
+                tx_hash_hex,
+                default_chain_spec.asdict(),
+                ],
+            queue=None,
+            )
+    t = s.apply_async()
+    r = t.get_leaf()
+
+    q = init_database.query(Otx)
+    q = q.filter(Otx.tx_hash==r)
+    otx = q.first()
+    if otx == None:
+        raise NotLocalTxError(r)
+
+    tx_after = unpack(bytes.fromhex(strip_0x(otx.signed_tx)), default_chain_spec.chain_id())
+    logg.debug('gasprices before {} after {}'.format(tx_before['gasPrice'], tx_after['gasPrice']))
+    assert tx_after['gasPrice'] > tx_before['gasPrice']
+

apps/cic-notify/requirements.txt

@@ -1,5 +1 @@
-alembic~=1.4.2
-celery~=4.4.7
-confini~=0.3.6rc3
-redis~=3.5.3
-semver==2.13.0
+cic_base[full_graph]~=0.1.2a46

apps/cic-ussd/cic_ussd/version.py

@@ -1,7 +1,7 @@
 # standard imports
 import semver
 
-version = (0, 3, 0, 'alpha.5')
+version = (0, 3, 0, 'alpha.7')
 
 version_object = semver.VersionInfo(
         major=version[0],

apps/cic-ussd/requirements.txt

@@ -1,49 +1,4 @@
-#cic_base[full_graph]~=0.1.1a24
-africastalking==1.2.3
-alembic==1.4.2
-amqp==2.6.1
-attrs==20.2.0
-bcrypt==3.2.0
-betterpath==0.2.2
-billiard==3.6.3.0
-celery==4.4.7
-cffi==1.14.3
+cic_base[full_graph]~=0.1.2a46
 cic-eth~=0.10.1b1
 cic-notify~=0.4.0a3
-cic-types~=0.1.0a8
-click==7.1.2
-confini~=0.3.6rc3
-cryptography==3.2.1
-faker==4.17.1
-iniconfig==1.1.1
-kombu==4.6.11
-Mako==1.1.3
-MarkupSafe==1.1.1
-mirakuru==2.3.0
-more-itertools==8.5.0
-packaging==20.4
-phonenumbers==8.12.12
-pluggy==0.13.1
-port-for==0.4
-psutil==5.7.3
-psycopg2==2.8.6
-py==1.9.0
-pycparser==2.20
-pyparsing==2.4.7
-python-dateutil==2.8.1
-python-editor==1.0.4
-python-gnupg==0.4.6
-python-i18n==0.3.9
-pytz==2020.1
-PyYAML==5.3.1
-redis==3.5.3
-semver==2.13.0
-six==1.15.0
-SQLAlchemy==1.3.20
-tinydb==4.2.0
-toml==0.10.1
-transitions==0.8.4
-uWSGI==2.0.19.1
-vcversioner==2.16.0.0
-vine==1.3.0
-zope.interface==5.1.2
+cic-types~=0.1.0a8

apps/contract-migration/reset.sh

@@ -40,7 +40,10 @@ if [[ -n "${ETH_PROVIDER}" ]]; then
 
 	#BANCOR_REGISTRY_ADDRESS=`cic-bancor-deploy --bancor-dir /usr/local/share/cic/bancor -z $DEV_ETH_RESERVE_ADDRESS -p $ETH_PROVIDER -o $DEV_ETH_ACCOUNT_CONTRACT_DEPLOYER`
 
+	>&2 echo "deploy account index contract"
 	DEV_ACCOUNT_INDEX_ADDRESS=`eth-accounts-index-deploy -i $CIC_CHAIN_SPEC -p $ETH_PROVIDER -y $keystore_file -vv -w`
+	>&2 echo "add deployer address as account index writer"
+	eth-accounts-index-writer -y $keystore_file -i $CIC_CHAIN_SPEC -p $ETH_PROVIDER -a $DEV_ACCOUNT_INDEX_ADDRESS -ww $debug $DEV_ETH_ACCOUNT_CONTRACT_DEPLOYER
 
 	CIC_REGISTRY_ADDRESS=`eth-contract-registry-deploy -i $CIC_CHAIN_SPEC -y $keystore_file --identifier BancorRegistry --identifier AccountRegistry --identifier TokenRegistry --identifier AddressDeclarator --identifier Faucet --identifier TransferAuthorization -p $ETH_PROVIDER -vv -w`
 	eth-contract-registry-set -w -y $keystore_file -r $CIC_REGISTRY_ADDRESS -i $CIC_CHAIN_SPEC  -p $ETH_PROVIDER -vv ContractRegistry $CIC_REGISTRY_ADDRESS

apps/contract-migration/scripts/README.md

@@ -21,9 +21,13 @@ Make sure the following is running in the cluster:
 	* eth
 	* postgres
 	* redis
+	* cic-meta-server
+
+
+If using the _custodial_ alternative for user imports, also run:
 	* cic-eth-tasker
 	* cic-eth-dispatcher
-	* cic-eth-manager-head
+	* cic-eth-tracker
 
 
 You will want to run these in sequence:
@@ -47,12 +51,25 @@ This will monitor new mined blocks and send balances to the newly created accoun
 
 ### 3. Users
 
+Only use **one** of the following
+
+#### Custodial
+
+This alternative generates accounts using the `cic-eth` custodial engine
+
 Without any modifications to the cluster and config files:
 
 `python import_users.py -c config --redis-host-callback redis <datadir>`
 
 ** A note on the The callback**:  The script uses a redis callback to retrieve the newly generated custodial address. This is the redis server _from the perspective of the cic-eth component_.
 
+#### Sovereign
+
+This alternative generates keystore files, while  registering corresponding addresses in the accounts registry directly
+
+`python import_sovereign_users.py -c config -i <newchain:id> -r <cic_registry_address> -p <eth_provider> -y ../keystore/UTC--2021-01-08T17-18-44.521011372Z--eb3907ecad74a0013c259d5874ae7f22dcbcc95c <datadir>`
+
+A `keystore` sub-directory in the data path is created, with ethereum keystore files for all generated private keys. Passphrase is set to empty string for all of them.
 
 ## VERIFY
 

apps/contract-migration/scripts/create_import_users.py

@@ -17,9 +17,7 @@ import random
 import vobject
 import celery
 from faker import Faker
-import cic_eth_registry
 import confini
-from cic_eth.api import Api
 from cic_types.models.person import (
         Person,
         generate_vcard_from_contact_data,
@@ -62,8 +60,6 @@ ts_then = int(dt_then.timestamp())
 
 celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'))
 
-api = Api(config.get('CIC_CHAIN_SPEC'))
-
 gift_max = args.gift_threshold or 0
 gift_factor = (10**6)
 

apps/contract-migration/scripts/import_sovereign_users.py

@@ -0,0 +1,181 @@
+# standard imports
+import os
+import sys
+import json
+import logging
+import argparse
+import uuid
+import datetime
+import time
+from glob import glob
+
+# external imports
+import confini
+from hexathon import (
+        add_0x,
+        strip_0x,
+        )
+from cic_types.models.person import Person
+from chainlib.eth.address import to_checksum_address
+from chainlib.chain import ChainSpec
+from chainlib.eth.connection import EthHTTPConnection
+from chainlib.eth.gas import RPCGasOracle
+from chainlib.eth.nonce import RPCNonceOracle
+from cic_types.processor import generate_metadata_pointer
+from eth_accounts_index import AccountRegistry
+from contract_registry import Registry
+from crypto_dev_signer.keystore.dict import DictKeystore
+from crypto_dev_signer.eth.signer.defaultsigner import ReferenceSigner as EIP155Signer
+from crypto_dev_signer.keystore.keyfile import to_dict as to_keyfile_dict
+
+logging.basicConfig(level=logging.WARNING)
+logg = logging.getLogger()
+
+default_config_dir = '/usr/local/etc/cic'
+
+argparser = argparse.ArgumentParser()
+argparser.add_argument('-p', '--provider', dest='p', default='http://localhost:8545', type=str, help='Web3 provider url (http only)')
+argparser.add_argument('-y', '--key-file', dest='y', type=str, help='Ethereum keystore file to use for signing')
+argparser.add_argument('-c', type=str, default=default_config_dir, help='config file')
+argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='Chain specification string')
+argparser.add_argument('-r', '--registry', dest='r', type=str, help='Contract registry address')
+argparser.add_argument('--batch-size', dest='batch_size', default=50, type=int, help='burst size of sending transactions to node')
+argparser.add_argument('--batch-delay', dest='batch_delay', default=2, type=int, help='seconds delay between batches')
+argparser.add_argument('-v', action='store_true', help='Be verbose')
+argparser.add_argument('-vv', action='store_true', help='Be more verbose')
+argparser.add_argument('user_dir', type=str, help='path to users export dir tree')
+args = argparser.parse_args()
+
+if args.v:
+    logg.setLevel(logging.INFO)
+elif args.vv:
+    logg.setLevel(logging.DEBUG)
+
+config_dir = args.c
+config = confini.Config(config_dir, os.environ.get('CONFINI_ENV_PREFIX'))
+config.process()
+args_override = {
+        'CIC_REGISTRY_ADDRESS': getattr(args, 'r'),
+        'CIC_CHAIN_SPEC': getattr(args, 'i'),
+        }
+config.dict_override(args_override, 'cli')
+config.add(args.user_dir, '_USERDIR', True)
+
+user_new_dir = os.path.join(args.user_dir, 'new')
+os.makedirs(user_new_dir)
+
+meta_dir = os.path.join(args.user_dir, 'meta')
+os.makedirs(meta_dir)
+
+user_old_dir = os.path.join(args.user_dir, 'old')
+os.stat(user_old_dir)
+
+chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
+chain_str = str(chain_spec)
+
+batch_size = args.batch_size
+batch_delay = args.batch_delay
+
+rpc = EthHTTPConnection(args.p)
+
+signer_address = None
+keystore = DictKeystore()
+if args.y != None:
+    logg.debug('loading keystore file {}'.format(args.y))
+    signer_address = keystore.import_keystore_file(args.y)
+    logg.debug('now have key for signer address {}'.format(signer_address))
+signer = EIP155Signer(keystore)
+
+nonce_oracle = RPCNonceOracle(signer_address, rpc)
+
+registry = Registry()
+o = registry.address_of(config.get('CIC_REGISTRY_ADDRESS'), 'AccountRegistry')
+r = rpc.do(o)
+account_registry_address = registry.parse_address_of(r)
+logg.info('using account registry {}'.format(account_registry_address))
+
+keyfile_dir = os.path.join(config.get('_USERDIR'), 'keystore')
+os.makedirs(keyfile_dir)
+
+def register_eth(i, u):
+
+    address_hex = keystore.new()
+    address = add_0x(to_checksum_address(address_hex))
+
+    gas_oracle = RPCGasOracle(rpc, code_callback=AccountRegistry.gas)
+    c = AccountRegistry(signer=signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle, chain_id=chain_spec.chain_id())
+    (tx_hash_hex, o) = c.add(account_registry_address, signer_address, address)
+    logg.debug('o {}'.format(o))
+    rpc.do(o)
+
+    pk = keystore.get(address)
+    keyfile_content = to_keyfile_dict(pk, '')
+    keyfile_path = os.path.join(keyfile_dir, '{}.json'.format(address))
+    f = open(keyfile_path, 'w')
+    json.dump(keyfile_content, f)
+    f.close()
+
+    logg.debug('[{}] register eth {} {} tx {} keyfile {}'.format(i, u, address, tx_hash_hex, keyfile_path))
+
+    return address
+   
+
+def register_ussd(u):
+    pass
+
+
+if __name__ == '__main__':
+
+
+    i = 0
+    j = 0
+    for x in os.walk(user_old_dir):
+        for y in x[2]:
+            if y[len(y)-5:] != '.json':
+                continue
+            filepath = os.path.join(x[0], y)
+            f = open(filepath, 'r')
+            try:
+                o = json.load(f)
+            except json.decoder.JSONDecodeError as e:
+                f.close()
+                logg.error('load error for {}: {}'.format(y, e))
+                continue
+            f.close()
+            u = Person.deserialize(o)
+
+            new_address = register_eth(i, u)
+            if u.identities.get('evm') == None:
+                u.identities['evm'] = {}
+            sub_chain_str = '{}:{}'.format(chain_spec.common_name(), chain_spec.network_id())
+            u.identities['evm'][sub_chain_str] = [new_address]
+
+            register_ussd(u)
+
+            new_address_clean = strip_0x(new_address)
+            filepath = os.path.join(
+                    user_new_dir,
+                    new_address_clean[:2].upper(),
+                    new_address_clean[2:4].upper(),
+                    new_address_clean.upper() + '.json',
+                    )
+            os.makedirs(os.path.dirname(filepath), exist_ok=True)
+
+            o = u.serialize()
+            f = open(filepath, 'w')
+            f.write(json.dumps(o))
+            f.close()
+
+            meta_key = generate_metadata_pointer(bytes.fromhex(new_address_clean), 'cic.person')
+            meta_filepath = os.path.join(meta_dir, '{}.json'.format(new_address_clean.upper()))
+            os.symlink(os.path.realpath(filepath), meta_filepath)
+
+            i += 1
+            sys.stdout.write('imported {} {}'.format(i, u).ljust(200) + "\r")
+        
+            j += 1
+            if j == batch_size:
+                time.sleep(batch_delay)
+                j = 0
+
+    #fi.close()

apps/contract-migration/scripts/import_users.py

@@ -36,7 +36,7 @@ argparser.add_argument('--redis-port', dest='redis_port', type=int, help='redis
 argparser.add_argument('--redis-db', dest='redis_db', type=int, help='redis db to use for task submission and callback')
 argparser.add_argument('--redis-host-callback', dest='redis_host_callback', default='localhost', type=str, help='redis host to use for callback')
 argparser.add_argument('--redis-port-callback', dest='redis_port_callback', default=6379, type=int, help='redis port to use for callback')
-argparser.add_argument('--batch-size', dest='batch_size', default=50, type=int, help='burst size of sending transactions to node')
+argparser.add_argument('--batch-size', dest='batch_size', default=80, type=int, help='burst size of sending transactions to node')
 argparser.add_argument('--batch-delay', dest='batch_delay', default=2, type=int, help='seconds delay between batches')
 argparser.add_argument('--timeout', default=60.0, type=float, help='Callback timeout')
 argparser.add_argument('-q', type=str, default='cic-eth', help='Task queue')

apps/contract-migration/scripts/requirements.txt

@@ -1,3 +1,5 @@
-cic-base[full_graph]==0.1.2a40
-cic-eth==0.10.1b1
+cic-base[full_graph]==0.1.2a52
+sarafu-faucet==0.0.2a15
+cic-eth==0.10.1b2+build.9c750e70
 cic-types==0.1.0a8
+crypto-dev-signer==0.4.14a17

apps/contract-migration/scripts/verify.py

@@ -10,6 +10,7 @@ import hashlib
 import csv
 import json
 import urllib
+import copy
 
 # external imports
 import celery
@@ -32,27 +33,45 @@ from chainlib.eth.block import (
 from chainlib.eth.hash import keccak256_string_to_hex
 from chainlib.eth.address import to_checksum_address
 from chainlib.eth.erc20 import ERC20
-from chainlib.eth.gas import OverrideGasOracle
+from chainlib.eth.gas import (
+        OverrideGasOracle,
+        balance,
+        )
 from chainlib.eth.tx import TxFactory
 from chainlib.eth.rpc import jsonrpc_template
 from chainlib.eth.error import EthException
-from cic_eth.api.api_admin import AdminApi
 from cic_types.models.person import (
         Person,
         generate_metadata_pointer,
         )
+from erc20_single_shot_faucet import SingleShotFaucet
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
 
 config_dir = '/usr/local/etc/cic-syncer'
 
+custodial_tests = [
+        'local_key',
+        'gas',
+        'faucet',
+        ]
+
+all_tests = custodial_tests + [
+        'accounts_index',
+        'balance',
+        'metadata',
+        ]
+
 argparser = argparse.ArgumentParser(description='daemon that monitors transactions in new blocks')
 argparser.add_argument('-p', '--provider', dest='p', type=str, help='chain rpc provider address')
 argparser.add_argument('-c', type=str, default=config_dir, help='config root to use')
 argparser.add_argument('--old-chain-spec', type=str, dest='old_chain_spec', default='evm:oldchain:1', help='chain spec')
 argparser.add_argument('-i', '--chain-spec', type=str, dest='i', help='chain spec')
 argparser.add_argument('--meta-provider', type=str, dest='meta_provider', default='http://localhost:63380', help='cic-meta url')
+argparser.add_argument('--skip-custodial', dest='skip_custodial', action='store_true', help='skip all custodial verifications')
+argparser.add_argument('--exclude', action='append', type=str, default=[], help='skip specified verification')
+argparser.add_argument('--include', action='append', type=str, help='include specified verification')
 argparser.add_argument('-r', '--registry-address', type=str, dest='r', help='CIC Registry address')
 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('-x', '--exit-on-error', dest='x', action='store_true', help='Halt exection on error')
@@ -91,14 +110,61 @@ user_dir = args.user_dir # user_out_dir from import_users.py
 meta_url = args.meta_provider
 exit_on_error = args.x
 
+active_tests = []
+exclude = []
+include = args.include
+if args.include == None:
+    include = all_tests
+for t in args.exclude:
+    if t not in all_tests:
+        raise ValueError('Cannot exclude unknown verification "{}"'.format(t))
+    exclude.append(t)
+if args.skip_custodial:
+    logg.info('will skip all custodial verifications ({})'.format(','.join(custodial_tests)))
+    for t in custodial_tests:
+        if t not in exclude:
+            exclude.append(t)
+for t in include:
+    if t not in all_tests:
+        raise ValueError('Cannot include unknown verification "{}"'.format(t))
+    if t not in exclude:
+        active_tests.append(t)
+        logg.info('will perform verification "{}"'.format(t))
+
+api = None
+for t in custodial_tests:
+    if t in active_tests:
+        from cic_eth.api.api_admin import AdminApi
+        api = AdminApi(None)
+        logg.info('activating custodial module'.format(t))
+        break
+
+cols = os.get_terminal_size().columns
+
+
+def to_terminalwidth(s):
+    ss = s.ljust(int(cols)-1)
+    ss += "\r"
+    return ss
+
+def default_outfunc(s):
+    ss = to_terminalwidth(s)
+    sys.stdout.write(ss)
+outfunc = default_outfunc
+if logg.isEnabledFor(logging.DEBUG):
+    outfunc = logg.debug
+
 
 class VerifierState:
 
-    def __init__(self, item_keys):
+    def __init__(self, item_keys, active_tests=None):
         self.items = {}
         for k in item_keys:
-            logg.info('k {}'.format(k))
             self.items[k] = 0
+        if active_tests == None:
+            self.active_tests = copy.copy(item_keys)
+        else:
+            self.active_tests = copy.copy(active_tests)
 
 
     def poke(self, item_key):
@@ -108,7 +174,10 @@ class VerifierState:
     def __str__(self):
         r = ''
         for k in self.items.keys():
-            r += '{}: {}\n'.format(k, self.items[k])
+            if k in self.active_tests:
+                r += '{}: {}\n'.format(k, self.items[k])
+            else:
+                r += '{}: skipped\n'.format(k)
         return r
 
 
@@ -127,25 +196,27 @@ class VerifierError(Exception):
 class Verifier:
 
     # TODO: what an awful function signature
-    def __init__(self, conn, cic_eth_api, gas_oracle, chain_spec, index_address, token_address, data_dir, exit_on_error=False):
+    def __init__(self, conn, cic_eth_api, gas_oracle, chain_spec, index_address, token_address, faucet_address, data_dir, exit_on_error=False):
         self.conn = conn
         self.gas_oracle = gas_oracle
         self.chain_spec = chain_spec
         self.index_address = index_address
         self.token_address = token_address
+        self.faucet_address = faucet_address
         self.erc20_tx_factory = ERC20(chain_id=chain_spec.chain_id(), gas_oracle=gas_oracle)
         self.tx_factory = TxFactory(chain_id=chain_spec.chain_id(), gas_oracle=gas_oracle)
         self.api = cic_eth_api
         self.data_dir = data_dir
         self.exit_on_error = exit_on_error
+        self.faucet_tx_factory = SingleShotFaucet(chain_id=chain_spec.chain_id(), gas_oracle=gas_oracle)
 
         verifymethods = []
         for k in dir(self):
             if len(k) > 7 and k[:7] == 'verify_':
-                logg.info('adding verify method {}'.format(k))
+                logg.debug('verifier has verify method {}'.format(k))
                 verifymethods.append(k[7:])
 
-        self.state = VerifierState(verifymethods)
+        self.state = VerifierState(verifymethods, active_tests=active_tests)
 
 
     def verify_accounts_index(self, address, balance=None):
@@ -182,6 +253,21 @@ class Verifier:
             raise VerifierError((address, r), 'local key')
 
 
+    def verify_gas(self, address, balance_token=None):
+        o = balance(address)
+        r = self.conn.do(o)
+        actual_balance = int(strip_0x(r), 16)
+        if actual_balance == 0:
+            raise VerifierError((address, actual_balance), 'gas')
+
+
+    def verify_faucet(self, address, balance_token=None):
+        o = self.faucet_tx_factory.usable_for(self.faucet_address, address)
+        r = self.conn.do(o)
+        if self.faucet_tx_factory.parse_usable_for(r):
+            raise VerifierError((address, r), 'faucet')
+
+
     def verify_metadata(self, address, balance=None):
         k = generate_metadata_pointer(bytes.fromhex(strip_0x(address)), ':cic.person')
         url = os.path.join(meta_url, k)
@@ -212,24 +298,14 @@ class Verifier:
             raise VerifierError(o_retrieved, 'metadata (person)')
 
 
-    def verify(self, address, balance):
-        logg.debug('verify {} {}'.format(address, balance))
+    def verify(self, address, balance, debug_stem=None):
   
-        methods = [
-                'local_key',
-                'accounts_index',
-                'balance',
-                'metadata',
-                ]
-
-        for k in methods:
+        for k in active_tests:
+            s = '{} {}'.format(debug_stem, k)
+            outfunc(s)
             try:
                 m = getattr(self, 'verify_{}'.format(k))
                 m(address, balance)
-#            self.verify_local_key(address)
-#            self.verify_accounts_index(address)
-#            self.verify_balance(address, balance)
-#            self.verify_metadata(address)
             except VerifierError as e:
                 logline = 'verification {} failed for {}: {}'.format(k, address, str(e))
                 if self.exit_on_error:
@@ -243,10 +319,6 @@ class Verifier:
         return str(self.state)
 
 
-class MockClient:
-
-    w3 = None
-
 def main():
     global chain_str, block_offset, user_dir
     
@@ -257,6 +329,7 @@ def main():
     txf = TxFactory(signer=None, gas_oracle=gas_oracle, nonce_oracle=None, chain_id=chain_spec.chain_id())
     tx = txf.template(ZERO_ADDRESS, config.get('CIC_REGISTRY_ADDRESS'))
 
+    # TODO: replace with cic-eth-registry
     registry_addressof_method = keccak256_string_to_hex('addressOf(bytes32)')[:8]
     data = add_0x(registry_addressof_method)
     data += eth_abi.encode_single('bytes32', b'TokenRegistry').hex()
@@ -267,7 +340,6 @@ def main():
     o['params'].append(txf.normalize(tx))
     o['params'].append('latest')
     r = conn.do(o)
-    print('r {}'.format(r))
     token_index_address = to_checksum_address(eth_abi.decode_single('address', bytes.fromhex(strip_0x(r))))
     logg.info('found token index address {}'.format(token_index_address))
 
@@ -283,6 +355,19 @@ def main():
     account_index_address = to_checksum_address(eth_abi.decode_single('address', bytes.fromhex(strip_0x(r))))
     logg.info('found account index address {}'.format(account_index_address))
 
+    data = add_0x(registry_addressof_method)
+    data += eth_abi.encode_single('bytes32', b'Faucet').hex()
+    txf.set_code(tx, data)
+    
+    o = jsonrpc_template()
+    o['method'] = 'eth_call'
+    o['params'].append(txf.normalize(tx))
+    o['params'].append('latest')
+    r = conn.do(o)
+    faucet_address = to_checksum_address(eth_abi.decode_single('address', bytes.fromhex(strip_0x(r))))
+    logg.info('found faucet {}'.format(faucet_address))
+
+
 
     # Get Sarafu token address
     tx = txf.template(ZERO_ADDRESS, token_index_address)
@@ -297,7 +382,6 @@ def main():
     o['params'].append(txf.normalize(tx))
     o['params'].append('latest')
     r = conn.do(o)
-    print('r {}'.format(r))
     sarafu_token_address = to_checksum_address(eth_abi.decode_single('address', bytes.fromhex(strip_0x(r))))
     logg.info('found token address {}'.format(sarafu_token_address))
 
@@ -312,7 +396,7 @@ def main():
         try:
             address = to_checksum_address(r[0])
             #sys.stdout.write('loading balance {} {}'.format(i, address).ljust(200) + "\r")
-            logg.debug('loading balance {} {}'.format(i, address).ljust(200))
+            outfunc('loading balance {} {}'.format(i, address)) #.ljust(200))
         except ValueError:
             break
         balance = int(r[1].rstrip())
@@ -321,11 +405,10 @@ def main():
 
     f.close()
 
-    api = AdminApi(MockClient())
-
-    verifier = Verifier(conn, api, gas_oracle, chain_spec, account_index_address, sarafu_token_address, user_dir, exit_on_error)
+    verifier = Verifier(conn, api, gas_oracle, chain_spec, account_index_address, sarafu_token_address, faucet_address, user_dir, exit_on_error)
 
     user_new_dir = os.path.join(user_dir, 'new')
+    i = 0
     for x in os.walk(user_new_dir):
         for y in x[2]:
             if y[len(y)-5:] != '.json':
@@ -341,7 +424,7 @@ def main():
             f.close()
 
             u = Person.deserialize(o)
-            logg.debug('data {}'.format(u.identities['evm']))
+            #logg.debug('data {}'.format(u.identities['evm']))
 
             subchain_str = '{}:{}'.format(chain_spec.common_name(), chain_spec.network_id())
             new_address = u.identities['evm'][subchain_str][0]
@@ -352,9 +435,11 @@ def main():
                 balance = balances[old_address]
             except KeyError:
                 logg.info('no old balance found for {}, assuming 0'.format(old_address))
-            logg.debug('checking {} -> {} = {}'.format(old_address, new_address, balance))
 
-            verifier.verify(new_address, balance)
+            s = 'checking {}: {} -> {} = {}'.format(i, old_address, new_address, balance)
+
+            verifier.verify(new_address, balance, debug_stem=s)
+            i += 1
 
     print(verifier)
 

apps/contract-migration/seed_cic_eth.sh

@@ -22,7 +22,7 @@ abi_dir=${ETH_ABI_DIR:-/usr/local/share/cic/solidity/abi}
 gas_amount=100000000000000000000000
 token_amount=${gas_amount}
 #faucet_amount=1000000000
-faucet_amount=0
+faucet_amount=${DEV_FAUCET_AMOUNT:-0}
 env_out_file=${CIC_DATA_DIR}/.env_seed
 init_level_file=${CIC_DATA_DIR}/.init
 truncate $env_out_file -s 0
@@ -32,11 +32,11 @@ set -a
 
 # get required addresses from registries
 DEV_TOKEN_INDEX_ADDRESS=`eth-contract-registry-list -i $CIC_CHAIN_SPEC -p $ETH_PROVIDER -r $CIC_REGISTRY_ADDRESS -f brief TokenRegistry`
-DEV_ACCOUNTS_INDEX_ADDRESS=`eth-contract-registry-list -i $CIC_CHAIN_SPEC -p $ETH_PROVIDER -r $CIC_REGISTRY_ADDRESS -f brief AccountRegistry`
+DEV_ACCOUNT_INDEX_ADDRESS=`eth-contract-registry-list -i $CIC_CHAIN_SPEC -p $ETH_PROVIDER -r $CIC_REGISTRY_ADDRESS -f brief AccountRegistry`
 DEV_RESERVE_ADDRESS=`eth-token-index-list -i $CIC_CHAIN_SPEC -p $ETH_PROVIDER -a $DEV_TOKEN_INDEX_ADDRESS -f brief SRF`
 cat <<EOF
 Token registry: $DEV_TOKEN_INDEX_ADDRESS
-Account reigstry: $DEV_ACCOUNTS_INDEX_ADDRESS
+Account reigstry: $DEV_ACCOUNT_INDEX_ADDRESS
 Reserve address: $DEV_RESERVE_ADDRESS
 EOF
 
@@ -66,7 +66,7 @@ DEV_ETH_ACCOUNT_ACCOUNT_REGISTRY_WRITER=`cic-eth-create $debug --redis-host-call
 echo DEV_ETH_ACCOUNT_ACCOUNT_REGISTRY_WRITER=$DEV_ETH_ACCOUNT_ACCOUNT_REGISTRY_WRITER >> $env_out_file
 cic-eth-tag -i $CIC_CHAIN_SPEC ACCOUNT_REGISTRY_WRITER $DEV_ETH_ACCOUNT_ACCOUNT_REGISTRY_WRITER
 >&2 echo "add acccounts index writer account as writer on contract"
-eth-accounts-index-writer -y $keystore_file -i $CIC_CHAIN_SPEC -p $ETH_PROVIDER -a $DEV_ACCOUNTS_INDEX_ADDRESS -ww $debug $DEV_ETH_ACCOUNT_ACCOUNT_REGISTRY_WRITER
+eth-accounts-index-writer -y $keystore_file -i $CIC_CHAIN_SPEC -p $ETH_PROVIDER -a $DEV_ACCOUNT_INDEX_ADDRESS -ww $debug $DEV_ETH_ACCOUNT_ACCOUNT_REGISTRY_WRITER
 
 # Transfer gas to custodial gas provider adddress
 >&2 echo gift gas to gas gifter

docker-compose.yml

@@ -102,6 +102,7 @@ services:
       CELERY_RESULT_URL: ${CELERY_RESULT_URL:-redis://redis:6379}
       DEV_PIP_EXTRA_INDEX_URL: ${DEV_PIP_EXTRA_INDEX_URL:-https://pip.grassrootseconomics.net:8433}
       RUN_MASK: ${RUN_MASK:-0} # bit flags; 1: contract migrations 2: seed data
+      DEV_FAUCET_AMOUNT: ${DEV_FAUCET_AMOUNT:-0}
     command: ["./run_job.sh"]
         #command: ["./reset.sh"]
     depends_on:
@@ -144,7 +145,7 @@ services:
       - -c
       - |
         if [[ -f /tmp/cic/config/.env ]]; then source /tmp/cic/config/.env; fi 
-        /usr/local/bin/cic-cache-trackerd -vv
+        /usr/local/bin/cic-cache-trackerd -vv -c /usr/local/etc/cic-cache
     volumes:
       - contract-config:/tmp/cic/config/:ro
 
@@ -316,8 +317,8 @@ services:
       CELERY_BROKER_URL: ${CELERY_BROKER_URL:-redis://redis}
       CELERY_RESULT_URL: ${CELERY_RESULT_URL:-redis://redis}
       TASKS_TRANSFER_CALLBACKS: $TASKS_TRANSFER_CALLBACKS
-        #DATABASE_DEBUG: ${DATABASE_DEBUG:-false}
-      DATABASE_DEBUG: 1
+      DATABASE_DEBUG: ${DATABASE_DEBUG:-false}
+        #DATABASE_DEBUG: 1
 
     depends_on:
       - eth
@@ -358,6 +359,8 @@ services:
       CELERY_RESULT_URL: ${CELERY_RESULT_URL:-redis://redis}
       TASKS_TRANSFER_CALLBACKS: $TASKS_TRANSFER_CALLBACKS
       CIC_TX_RETRY_DELAY: 15
+      BATCH_SIZE: ${RETRIER_BATCH_SIZE:-50}
+        #DATABASE_DEBUG: 1
     depends_on:
       - eth
       - postgres
@@ -372,7 +375,7 @@ services:
       - -c
       - |
         if [[ -f /tmp/cic/config/.env ]]; then source /tmp/cic/config/.env; fi 
-        ./start_retry.sh -v
+        ./start_retry.sh -vv 
     # command: "/root/start_retry.sh -q cic-eth -vv"