Merge branch 'master' into lash/stale-import-deps

Fix false offset limit error

Closes #60

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

.gitignore

@@ -1,2 +1,10 @@
 service-configs/*
 !service-configs/.gitkeep
+**/node_modules/
+__pycache__
+*.pyc
+*.o
+gmon.out
+*.egg-info
+dist/
+build/

.gitlab-ci.yml

@@ -6,6 +6,7 @@ include:
   - local: 'apps/cic-notify/.gitlab-ci.yml'
   - local: 'apps/cic-meta/.gitlab-ci.yml'
   - local: 'apps/cic-cache/.gitlab-ci.yml'
+  - local: 'apps/data-seeding/.gitlab-ci.yml'
 
 stages:
   - build

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

@@ -6,3 +6,4 @@ HOST=localhost
 PORT=5432
 ENGINE=postgresql
 DRIVER=psycopg2
+DEBUG=

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

@@ -6,3 +6,4 @@ HOST=localhost
 PORT=5432
 ENGINE=sqlite
 DRIVER=pysqlite
+DEBUG=

apps/cic-cache/cic_cache/cache.py

@@ -1,22 +1,28 @@
 # standard imports
 import logging
+import datetime
 
-# third-party imports
+# external imports
 import moolb
 
 # local imports
-from cic_cache.db import list_transactions_mined
-from cic_cache.db import list_transactions_account_mined
+from cic_cache.db.list import (
+        list_transactions_mined,
+        list_transactions_account_mined,
+        list_transactions_mined_with_data,
+        )
 
 logg = logging.getLogger()
 
 
-class BloomCache:
+class Cache:
 
     def __init__(self, session):
         self.session = session
 
 
+class BloomCache(Cache):
+
     @staticmethod
     def __get_filter_size(n):
         n = 8192 * 8
@@ -87,3 +93,44 @@ class BloomCache:
             f_blocktx.add(block + tx)
             logg.debug('added block {} tx {} lo {} hi {}'.format(r[0], r[1], lowest_block, highest_block))
         return (lowest_block, highest_block, f_block.to_bytes(), f_blocktx.to_bytes(),)
+
+
+class DataCache(Cache):
+
+    def load_transactions_with_data(self, offset, end):
+        rows = list_transactions_mined_with_data(self.session, offset, end) 
+        tx_cache = []
+        highest_block = -1;
+        lowest_block = -1;
+        date_is_str = None # stick this in startup
+        for r in rows:
+            if highest_block == -1:
+                highest_block = r['block_number']
+            lowest_block = r['block_number']
+            tx_type = 'unknown'
+
+            if r['value'] != None:
+                tx_type = '{}.{}'.format(r['domain'], r['value'])
+
+            if date_is_str == None:
+                date_is_str = type(r['date_block']).__name__ == 'str'
+
+            o = {
+                'block_number': r['block_number'],
+                'tx_hash': r['tx_hash'],
+                'date_block': r['date_block'],
+                'sender': r['sender'],
+                'recipient': r['recipient'],
+                'from_value': int(r['from_value']),
+                'to_value': int(r['to_value']),
+                'source_token': r['source_token'],
+                'destination_token': r['destination_token'],
+                'success': r['success'],
+                'tx_type': tx_type,
+            }
+
+            if date_is_str:
+                o['date_block'] = datetime.datetime.fromisoformat(r['date_block'])
+
+            tx_cache.append(o)
+        return (lowest_block, highest_block, tx_cache)

apps/cic-cache/cic_cache/db/__init__.py

@@ -2,9 +2,14 @@
 import logging
 
 # local imports
-from .list import list_transactions_mined
-from .list import list_transactions_account_mined
-from .list import add_transaction
+from .list import (
+        list_transactions_mined,
+        list_transactions_account_mined,
+        add_transaction,
+        tag_transaction,
+        add_tag,
+    )
+
 
 logg = logging.getLogger()
 

apps/cic-cache/cic_cache/db/list.py

@@ -2,8 +2,9 @@
 import logging
 import datetime
 
-# third-party imports
+# external imports
 from cic_cache.db.models.base import SessionBase
+from sqlalchemy import text
 
 logg = logging.getLogger()
 
@@ -27,6 +28,26 @@ def list_transactions_mined(
     return r
 
 
+def list_transactions_mined_with_data(
+        session,
+        offset,
+        end,
+        ):
+    """Executes db query to return all confirmed transactions according to the specified offset and limit.
+
+    :param offset: Offset in data set to return transactions from
+    :type offset: int
+    :param limit: Max number of transactions to retrieve
+    :type limit: int
+    :result: Result set
+    :rtype: SQLAlchemy.ResultProxy
+    """
+    s = "SELECT tx_hash, block_number, date_block, sender, recipient, from_value, to_value, source_token, destination_token, success, domain, value FROM tx LEFT JOIN tag_tx_link ON tx.id = tag_tx_link.tx_id LEFT JOIN tag ON tag_tx_link.tag_id = tag.id WHERE block_number >= {} AND block_number <= {} ORDER BY block_number ASC, tx_index ASC".format(offset, end)
+
+    r = session.execute(s)
+    return r
+
+
 def list_transactions_account_mined(
         session,
         address,
@@ -50,7 +71,8 @@ def list_transactions_account_mined(
 
 
 def add_transaction(
-        session, tx_hash,
+        session,
+        tx_hash,
         block_number,
         tx_index,
         sender,
@@ -62,6 +84,33 @@ def add_transaction(
         success,
         timestamp,
         ):
+    """Adds a single transaction to the cache persistent storage. Sensible interpretation of all fields is the responsibility of the caller.
+
+    :param session: Persistent storage session object
+    :type session: SQLAlchemy session
+    :param tx_hash: Transaction hash
+    :type tx_hash: str, 0x-hex
+    :param block_number: Block number
+    :type block_number: int
+    :param tx_index: Transaction index in block
+    :type tx_index: int
+    :param sender: Ethereum address of effective sender
+    :type sender: str, 0x-hex
+    :param receiver: Ethereum address of effective recipient
+    :type receiver: str, 0x-hex
+    :param source_token: Ethereum address of token used by sender
+    :type source_token: str, 0x-hex
+    :param destination_token: Ethereum address of token received by recipient
+    :type destination_token: str, 0x-hex
+    :param from_value: Source token value spent in transaction
+    :type from_value: int
+    :param to_value: Destination token value received in transaction
+    :type to_value: int
+    :param success: True if code execution on network was successful
+    :type success: bool
+    :param date_block: Block timestamp
+    :type date_block: datetime
+    """
     date_block = datetime.datetime.fromtimestamp(timestamp)
     s = "INSERT INTO tx (tx_hash, block_number, tx_index, sender, recipient, source_token, destination_token, from_value, to_value, success, date_block) VALUES ('{}', {}, {}, '{}', '{}', '{}', '{}', {}, {}, {}, '{}')".format(
             tx_hash,
@@ -77,3 +126,74 @@ def add_transaction(
             date_block,
             )
     session.execute(s)
+
+
+
+def tag_transaction(
+        session,
+        tx_hash,
+        name,
+        domain=None,
+        ):
+    """Tag a single transaction with a single tag.
+
+    Tag must already exist in storage.
+
+    :param session: Persistent storage session object
+    :type session: SQLAlchemy session
+    :param tx_hash: Transaction hash
+    :type tx_hash: str, 0x-hex
+    :param name: Tag value
+    :type name: str
+    :param domain: Tag domain
+    :type domain: str
+    :raises ValueError: Unknown tag or transaction hash
+
+    """
+
+    s = text("SELECT id from tx where tx_hash = :a")
+    r = session.execute(s, {'a': tx_hash}).fetchall()
+    tx_id = r[0].values()[0]
+
+    if tx_id == None:
+        raise ValueError('unknown tx hash {}'.format(tx_hash))
+
+    #s = text("SELECT id from tag where value = :a and domain = :b")
+    if domain == None:
+        s = text("SELECT id from tag where value = :a")
+    else:
+        s = text("SELECT id from tag where value = :a and domain = :b")
+    r = session.execute(s, {'a': name, 'b': domain}).fetchall()
+    tag_id = r[0].values()[0]
+
+    logg.debug('type {} {}'.format(type(tag_id), type(tx_id)))
+
+    if tag_id == None:
+        raise ValueError('unknown tag name {} domain {}'.format(name, domain))
+
+    s = text("INSERT INTO tag_tx_link (tag_id, tx_id) VALUES (:a, :b)")
+    r = session.execute(s, {'a': int(tag_id), 'b': int(tx_id)})
+
+
+def add_tag(
+        session,
+        name,
+        domain=None,
+        ):
+    """Add a single tag to storage.
+
+    :param session: Persistent storage session object
+    :type session: SQLAlchemy session
+    :param name: Tag value
+    :type name: str
+    :param domain: Tag domain
+    :type domain: str
+    :raises sqlalchemy.exc.IntegrityError: Tag already exists
+    """
+
+    s = None
+    if domain == None: 
+        s = text("INSERT INTO tag (value) VALUES (:b)")
+    else:
+        s = text("INSERT INTO tag (domain, value) VALUES (:a, :b)")
+    session.execute(s, {'a': domain, 'b': name})

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

@@ -0,0 +1,38 @@
+"""Transaction tags
+
+Revision ID: aaf2bdce7d6e
+Revises: 6604de4203e2
+Create Date: 2021-05-01 09:20:20.775082
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'aaf2bdce7d6e'
+down_revision = '6604de4203e2'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+            'tag',
+            sa.Column('id', sa.Integer, primary_key=True),
+            sa.Column('domain', sa.String(), nullable=True),
+            sa.Column('value', sa.String(), nullable=False),
+            )
+    op.create_index('idx_tag_domain_value', 'tag', ['domain', 'value'], unique=True)
+
+    op.create_table(
+            'tag_tx_link',
+            sa.Column('id', sa.Integer, primary_key=True),
+            sa.Column('tag_id', sa.Integer, sa.ForeignKey('tag.id'), nullable=False),
+            sa.Column('tx_id', sa.Integer, sa.ForeignKey('tx.id'), nullable=False),
+                )
+
+def downgrade():
+    op.drop_table('tag_tx_link')
+    op.drop_index('idx_tag_domain_value')
+    op.drop_table('tag')

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

@@ -1 +1,2 @@
 from .erc20 import *
+from .faucet import *

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

@@ -1,2 +1,27 @@
-class SyncFilter:
-    pass
+class TagSyncFilter:
+    """Holds tag name and domain for an implementing filter.
+
+    :param name: Tag value 
+    :type name: str
+    :param domain: Tag domain
+    :type domain: str
+    """
+
+    def __init__(self, name, domain=None):
+        self.tag_name = name
+        self.tag_domain = domain
+
+
+    def tag(self):
+        """Return tag value/domain.
+
+        :rtype: Tuple
+        :returns: tag value/domain.
+        """
+        return (self.tag_name, self.tag_domain)
+
+
+    def __str__(self):
+        if self.tag_domain == None:
+            return self.tag_name
+        return '{}.{}'.format(self.tag_domain, self.tag_name)

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

@@ -2,7 +2,6 @@
 import logging
 
 # external imports
-from chainlib.eth.erc20 import ERC20
 from chainlib.eth.address import (
         to_checksum_address,
         )
@@ -13,17 +12,19 @@ from cic_eth_registry.error import (
         NotAContractError,
         ContractMismatchError,
         )
+from eth_erc20 import ERC20
 
 # local imports
-from .base import SyncFilter
+from .base import TagSyncFilter
 from cic_cache import db as cic_cache_db
 
 logg = logging.getLogger().getChild(__name__)
 
 
-class ERC20TransferFilter(SyncFilter):
+class ERC20TransferFilter(TagSyncFilter):
 
     def __init__(self, chain_spec):
+        super(ERC20TransferFilter, self).__init__('transfer', domain='erc20')
         self.chain_spec = chain_spec
 
 
@@ -32,7 +33,7 @@ class ERC20TransferFilter(SyncFilter):
         logg.debug('filter {} {}'.format(block, tx))
         token = None
         try:
-            token = ERC20Token(conn, tx.inputs[0])
+            token = ERC20Token(self.chain_spec, conn, tx.inputs[0])
         except NotAContractError:
             logg.debug('not a contract {}'.format(tx.inputs[0]))
             return False
@@ -46,6 +47,9 @@ class ERC20TransferFilter(SyncFilter):
         except RequestMismatchException:
             logg.debug('erc20 match but not a transfer, skipping')
             return False
+        except ValueError:
+            logg.debug('erc20 match but bogus data, skipping')
+            return False
 
         token_sender = tx.outputs[0]
         token_recipient = transfer_data[0]
@@ -68,5 +72,12 @@ class ERC20TransferFilter(SyncFilter):
                 block.timestamp,
                 )
         db_session.flush()
+        cic_cache_db.tag_transaction(
+                db_session,
+                tx.hash,
+                self.tag_name,
+                domain=self.tag_domain,
+                )
+        db_session.commit()
 
         return True

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

@@ -0,0 +1,73 @@
+# standard imports
+import logging
+
+# external imports
+from erc20_faucet import Faucet
+from chainlib.eth.address import to_checksum_address
+from chainlib.eth.constant import ZERO_ADDRESS
+from chainlib.status import Status
+from hexathon import strip_0x
+
+# local imports
+import cic_cache.db as cic_cache_db
+from .base import TagSyncFilter
+
+#logg = logging.getLogger().getChild(__name__)
+logg = logging.getLogger()
+
+
+class FaucetFilter(TagSyncFilter):
+
+    def __init__(self, chain_spec, sender_address=ZERO_ADDRESS):
+        super(FaucetFilter, self).__init__('give_to', domain='faucet')
+        self.chain_spec = chain_spec
+        self.sender_address = sender_address
+
+
+    def filter(self, conn, block, tx, db_session=None):
+        try:
+            data = strip_0x(tx.payload)
+        except ValueError:
+            return False
+        logg.debug('data {}'.format(data))
+        if Faucet.method_for(data[:8]) == None:
+            return False
+
+        token_sender = tx.inputs[0]
+        token_recipient = data[64+8-40:]
+        logg.debug('token recipient {}'.format(token_recipient))
+ 
+        f = Faucet(self.chain_spec)
+        o = f.token(token_sender, sender_address=self.sender_address)
+        r = conn.do(o)
+        token = f.parse_token(r)
+
+        f = Faucet(self.chain_spec)
+        o = f.token_amount(token_sender, sender_address=self.sender_address)
+        r = conn.do(o)
+        token_value = f.parse_token_amount(r)
+
+        cic_cache_db.add_transaction(
+                db_session,
+                tx.hash,
+                block.number,
+                tx.index,
+                to_checksum_address(token_sender),
+                to_checksum_address(token_recipient),
+                token,
+                token,
+                token_value,
+                token_value,
+                tx.status == Status.SUCCESS,
+                block.timestamp,
+                )
+        db_session.flush()
+        cic_cache_db.tag_transaction(
+                db_session,
+                tx.hash,
+                self.tag_name,
+                domain=self.tag_domain,
+                )
+        db_session.commit()
+
+        return True

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

@@ -0,0 +1,110 @@
+# standard imports
+import logging
+import json
+import re
+import base64
+
+# local imports
+from cic_cache.cache import (
+        BloomCache,
+        DataCache,
+    )
+
+logg = logging.getLogger(__name__)
+
+re_transactions_all_bloom = r'/tx/(\d+)?/?(\d+)/?'
+re_transactions_account_bloom = r'/tx/user/((0x)?[a-fA-F0-9]+)/?(\d+)?/?(\d+)/?'
+re_transactions_all_data = r'/txa/(\d+)/(\d+)/?'
+
+DEFAULT_LIMIT = 100
+
+
+def process_transactions_account_bloom(session, env):
+    r = re.match(re_transactions_account_bloom, env.get('PATH_INFO'))
+    if not r:
+        return None
+
+    address = r[1]
+    if r[2] == None:
+        address = '0x' + address
+    offset = DEFAULT_LIMIT
+    if r.lastindex > 2:
+        offset = r[3]
+    limit = 0
+    if r.lastindex > 3:
+        limit = r[4]
+
+    c = BloomCache(session)
+    (lowest_block, highest_block, bloom_filter_block, bloom_filter_tx) = c.load_transactions_account(address, offset, limit)
+
+    o = {
+        'alg': 'sha256',
+        'low': lowest_block,
+        'high': highest_block,
+        'block_filter': base64.b64encode(bloom_filter_block).decode('utf-8'),
+        'blocktx_filter': base64.b64encode(bloom_filter_tx).decode('utf-8'),
+        'filter_rounds': 3,
+            }
+
+    j = json.dumps(o)
+
+    return ('application/json', j.encode('utf-8'),)
+
+
+def process_transactions_all_bloom(session, env):
+    r = re.match(re_transactions_all_bloom, env.get('PATH_INFO'))
+    if not r:
+        return None
+
+    offset = DEFAULT_LIMIT
+    if r.lastindex > 0:
+        offset = r[1]
+    limit = 0
+    if r.lastindex > 1:
+        limit = r[2]
+
+    c = BloomCache(session)
+    (lowest_block, highest_block, bloom_filter_block, bloom_filter_tx) = c.load_transactions(offset, limit)
+
+    o = {
+        'alg': 'sha256',
+        'low': lowest_block,
+        'high': highest_block,
+        'block_filter': base64.b64encode(bloom_filter_block).decode('utf-8'),
+        'blocktx_filter': base64.b64encode(bloom_filter_tx).decode('utf-8'),
+        'filter_rounds': 3,
+            }
+
+    j = json.dumps(o)
+
+    return ('application/json', j.encode('utf-8'),)
+
+
+def process_transactions_all_data(session, env):
+    r = re.match(re_transactions_all_data, env.get('PATH_INFO'))
+    if not r:
+        return None
+    if env.get('HTTP_X_CIC_CACHE_MODE') != 'all':
+        return None
+
+    offset = r[1]
+    end = r[2]
+    if int(r[2]) < int(r[1]):
+        raise ValueError('cart before the horse, dude')
+
+    c = DataCache(session)
+    (lowest_block, highest_block, tx_cache) = c.load_transactions_with_data(offset, end)
+
+    for r in tx_cache:
+        r['date_block'] = r['date_block'].timestamp()
+
+    o = {
+        'low': lowest_block,
+        'high': highest_block,
+        'data': tx_cache,
+    }
+
+    
+    j = json.dumps(o)
+
+    return ('application/json', j.encode('utf-8'),)

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

@@ -1,18 +1,20 @@
 # standard imports
 import os
-import re
 import logging
 import argparse
-import json
 import base64
 
-# third-party imports
+# external imports
 import confini
 
 # local imports
-from cic_cache import BloomCache
 from cic_cache.db import dsn_from_config
 from cic_cache.db.models.base import SessionBase
+from cic_cache.runnable.daemons.query import (
+        process_transactions_account_bloom,
+        process_transactions_all_bloom,
+        process_transactions_all_data,
+        )
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -44,72 +46,6 @@ logg.debug('config:\n{}'.format(config))
 dsn = dsn_from_config(config)
 SessionBase.connect(dsn, config.true('DATABASE_DEBUG'))
 
-re_transactions_all_bloom = r'/tx/(\d+)?/?(\d+)/?'
-re_transactions_account_bloom = r'/tx/user/((0x)?[a-fA-F0-9]+)/?(\d+)?/?(\d+)/?'
-
-DEFAULT_LIMIT = 100
-
-
-def process_transactions_account_bloom(session, env):
-    r = re.match(re_transactions_account_bloom, env.get('PATH_INFO'))
-    if not r:
-        return None
-
-    address = r[1]
-    if r[2] == None:
-        address = '0x' + address
-    offset = DEFAULT_LIMIT
-    if r.lastindex > 2:
-        offset = r[3]
-    limit = 0
-    if r.lastindex > 3:
-        limit = r[4]
-
-    c = BloomCache(session)
-    (lowest_block, highest_block, bloom_filter_block, bloom_filter_tx) = c.load_transactions_account(address, offset, limit)
-
-    o = {
-        'alg': 'sha256',
-        'low': lowest_block,
-        'high': highest_block,
-        'block_filter': base64.b64encode(bloom_filter_block).decode('utf-8'),
-        'blocktx_filter': base64.b64encode(bloom_filter_tx).decode('utf-8'),
-        'filter_rounds': 3,
-            }
-
-    j = json.dumps(o)
-
-    return ('application/json', j.encode('utf-8'),)
-
-
-def process_transactions_all_bloom(session, env):
-    r = re.match(re_transactions_all_bloom, env.get('PATH_INFO'))
-    if not r:
-        return None
-
-    offset = DEFAULT_LIMIT
-    if r.lastindex > 0:
-        offset = r[1]
-    limit = 0
-    if r.lastindex > 1:
-        limit = r[2]
-
-    c = BloomCache(session)
-    (lowest_block, highest_block, bloom_filter_block, bloom_filter_tx) = c.load_transactions(offset, limit)
-
-    o = {
-        'alg': 'sha256',
-        'low': lowest_block,
-        'high': highest_block,
-        'block_filter': base64.b64encode(bloom_filter_block).decode('utf-8'),
-        'blocktx_filter': base64.b64encode(bloom_filter_tx).decode('utf-8'),
-        'filter_rounds': 3,
-            }
-
-    j = json.dumps(o)
-
-    return ('application/json', j.encode('utf-8'),)
-
 
 # uwsgi application
 def application(env, start_response):
@@ -119,10 +55,16 @@ def application(env, start_response):
 
     session = SessionBase.create_session()
     for handler in [
+            process_transactions_all_data,
             process_transactions_all_bloom,
             process_transactions_account_bloom,
             ]:
-        r = handler(session, env)
+        r = None
+        try:
+            r = handler(session, env)
+        except ValueError as e:
+            start_response('400 {}'.format(str(e)))
+            return []
         if r != None:
             (mime_type, content) = r
             break

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

@@ -7,9 +7,10 @@ import argparse
 import sys
 import re
 
-# third-party imports
+# external imports
 import confini
 import celery
+import sqlalchemy
 import rlp
 import cic_base.config
 import cic_base.log
@@ -26,26 +27,40 @@ from chainlib.eth.block import (
 from hexathon import (
         strip_0x,
         )
-from chainsyncer.backend import SyncerBackend
+from chainsyncer.backend.sql import SQLBackend
 from chainsyncer.driver import (
         HeadSyncer,
+        HistorySyncer,
         )
 from chainsyncer.db.models.base import SessionBase
 
 # local imports
-from cic_cache.db import dsn_from_config
+from cic_cache.db import (
+        dsn_from_config,
+        add_tag,
+        )
 from cic_cache.runnable.daemons.filters import (
         ERC20TransferFilter,
+        FaucetFilter,
         )
 
 script_dir = os.path.realpath(os.path.dirname(__file__))
 
+def add_block_args(argparser):
+    argparser.add_argument('--history-start', type=int, default=0, dest='history_start', help='Start block height for initial history sync')
+    argparser.add_argument('--no-history', action='store_true', dest='no_history', help='Skip initial history sync')
+    return argparser
+
+
 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')
+argparser = cic_base.argparse.add(argparser, add_block_args, 'block')
 args = cic_base.argparse.parse(argparser, logg)
 config = cic_base.config.create(args.c, args, args.env_prefix)
 
+config.add(args.history_start, 'SYNCER_HISTORY_START', True)
+config.add(args.no_history, '_NO_HISTORY', True)
+
 cic_base.config.log(config)
 
 dsn = dsn_from_config(config)
@@ -54,10 +69,21 @@ 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 register_filter_tags(filters, session):
+    for f in filters:
+        tag = f.tag()
+        try:
+            add_tag(session, tag[0], domain=tag[1])
+            session.commit()
+            logg.info('added tag name "{}" domain "{}"'.format(tag[0], tag[1]))
+        except sqlalchemy.exc.IntegrityError:
+            session.rollback()
+            logg.debug('already have tag name "{}" domain "{}"'.format(tag[0], tag[1]))
+
+
 def main():
     # Connect to blockchain with chainlib
     rpc = RPCConnection.connect(chain_spec, 'default')
@@ -66,23 +92,30 @@ def main():
     r = rpc.do(o)
     block_offset = int(strip_0x(r), 16) + 1
 
-    logg.debug('starting at block {}'.format(block_offset))
+    logg.debug('current block height {}'.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 SQLBackend.first(chain_spec):
+    #    backend = SQLBackend.initial(chain_spec, block_offset)
+    syncer_backends = SQLBackend.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))
+        initial_block_start = config.get('SYNCER_HISTORY_START')
+        initial_block_offset = block_offset
+        if config.get('_NO_HISTORY'):
+            initial_block_start = block_offset
+            initial_block_offset += 1
+        syncer_backends.append(SQLBackend.initial(chain_spec, initial_block_offset, start_block_height=initial_block_start))
+        logg.info('found no backends to resume, adding initial sync from history start {} end {}'.format(initial_block_start, initial_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))
+    for syncer_backend in syncer_backends:
+        syncers.append(HistorySyncer(syncer_backend))
 
+    syncer_backend = SQLBackend.live(chain_spec, block_offset+1)
     syncers.append(HeadSyncer(syncer_backend))
 
     trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
@@ -94,11 +127,22 @@ def main():
         logg.info('using trusted address {}'.format(address))
 
     erc20_transfer_filter = ERC20TransferFilter(chain_spec)
+    faucet_filter = FaucetFilter(chain_spec)
+
+    filters = [
+        erc20_transfer_filter,
+        faucet_filter,
+            ]
+
+    session = SessionBase.create_session()
+    register_filter_tags(filters, session)
+    session.close()
 
     i = 0
     for syncer in syncers:
         logg.debug('running syncer index {}'.format(i))
-        syncer.add_filter(erc20_transfer_filter)
+        for f in filters:
+            syncer.add_filter(f)
         r = syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')), rpc)
         sys.stderr.write("sync {} done at block {}\n".format(syncer, r))
 

apps/cic-cache/config/database.ini

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

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

@@ -1,2 +1,2 @@
 [eth]
-provider = ws://localhost:63546
+provider = http://localhost:63545

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

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

apps/cic-cache/config/syncer.ini

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

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

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

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

@@ -6,4 +6,4 @@ HOST=localhost
 PORT=5432
 ENGINE=sqlite
 DRIVER=pysqlite
-DEBUG=
+DEBUG=1

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

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

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.2a53
+RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2b9
 
 COPY cic-cache/requirements.txt ./
 COPY cic-cache/setup.cfg \
@@ -43,10 +43,9 @@ COPY cic-cache/config/ /usr/local/etc/cic-cache/
 RUN git clone https://github.com/vishnubob/wait-for-it.git /usr/local/bin/wait-for-it/
 COPY cic-cache/cic_cache/db/migrations/ /usr/local/share/cic-cache/alembic/
 
-RUN git clone https://gitlab.com/grassrootseconomics/cic-contracts.git && \
-	mkdir -p /usr/local/share/cic/solidity && \
-	cp -R cic-contracts/abis /usr/local/share/cic/solidity/abi
-
+COPY cic-cache/docker/start_tracker.sh ./start_tracker.sh
+COPY cic-cache/docker/db.sh ./db.sh
+RUN chmod 755 ./*.sh
 # Tracker
 # ENTRYPOINT ["/usr/local/bin/cic-cache-tracker", "-vv"]
 # Server

apps/cic-cache/docker/db.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+
+set -e
+>&2 echo executing database migration
+python scripts/migrate.py -c /usr/local/etc/cic-cache --migrations-dir /usr/local/share/cic-cache/alembic -vv
+set +e

apps/cic-cache/docker/start_tracker.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+. ./db.sh
+
+if [ $? -ne "0" ]; then
+	>&2 echo db migrate fail
+	exit 1
+fi
+
+/usr/local/bin/cic-cache-trackerd $@

apps/cic-cache/requirements.txt

@@ -1,13 +1,12 @@
-cic-base~=0.1.2a53
+cic-base~=0.1.2b10
 alembic==1.4.2
 confini~=0.3.6rc3
 uwsgi==2.0.19.1
 moolb~=0.1.0
-cic-eth-registry~=0.5.4a9
+cic-eth-registry~=0.5.5a4
 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
+chainsyncer[sql]~=0.0.2a4

apps/cic-cache/test_requirements.txt

@@ -4,3 +4,8 @@ pytest-mock==3.3.1
 pysqlite3==0.4.3
 sqlparse==0.4.1
 pytest-celery==0.0.0a1
+eth_tester==0.5.0b3
+py-evm==0.3.0a20
+web3==5.12.2
+cic-eth-registry~=0.5.5a3
+cic-base[full]==0.1.2b8

apps/cic-cache/tests/conftest.py

@@ -3,7 +3,7 @@ import os
 import sys
 import datetime
 
-# third-party imports
+# external imports
 import pytest
 
 # local imports
@@ -84,3 +84,20 @@ def txs(
     
     session.commit()
 
+    return [
+            tx_hash_first,
+            tx_hash_second,
+            ]
+
+
+@pytest.fixture(scope='function')
+def tag_txs(
+        init_database,
+        txs,
+        ):
+
+    db.add_tag(init_database, 'taag', domain='test')
+    init_database.commit()
+
+    db.tag_transaction(init_database, txs[1], 'taag', domain='test')
+

apps/cic-cache/tests/filters/conftest.py

@@ -0,0 +1,3 @@
+from chainlib.eth.pytest import *
+from cic_eth_registry.pytest.fixtures_tokens import *
+

apps/cic-cache/tests/filters/test_erc20.py

@@ -0,0 +1,69 @@
+# standard imports
+import os
+import datetime
+import logging
+import json
+
+# external imports
+import pytest
+from sqlalchemy import text
+from chainlib.eth.tx import Tx
+from chainlib.eth.block import Block
+from chainlib.chain import ChainSpec
+from hexathon import (
+        strip_0x,
+        add_0x,
+        )
+
+# local imports
+from cic_cache.db import add_tag
+from cic_cache.runnable.daemons.filters.erc20 import ERC20TransferFilter
+
+logg = logging.getLogger()
+
+
+def test_erc20_filter(
+        eth_rpc,
+        foo_token,
+        init_database,
+        list_defaults,
+        list_actors,
+        tags,
+        ):
+
+    chain_spec = ChainSpec('foo', 'bar', 42, 'baz')
+
+    fltr = ERC20TransferFilter(chain_spec)
+
+    add_tag(init_database, fltr.tag_name, domain=fltr.tag_domain)
+
+    data = 'a9059cbb'
+    data += strip_0x(list_actors['alice'])
+    data += '1000'.ljust(64, '0')
+
+    block = Block({
+        'hash': os.urandom(32).hex(),
+        'number': 42,
+        'timestamp': datetime.datetime.utcnow().timestamp(),
+        'transactions': [],
+        })
+
+    tx = Tx({
+        'to': foo_token,
+        'from': list_actors['bob'],
+        'data': data,
+        'value': 0,
+        'hash': os.urandom(32).hex(),
+        'nonce': 13,
+        'gasPrice': 10000000,
+        'gas': 123456,
+            })
+    block.txs.append(tx)
+    tx.block = block
+
+    r = fltr.filter(eth_rpc, block, tx, db_session=init_database)
+    assert r
+
+    s = text("SELECT x.tx_hash FROM tag a INNER JOIN tag_tx_link l ON l.tag_id = a.id INNER JOIN tx x ON x.id = l.tx_id WHERE a.domain = :a AND a.value = :b")
+    r = init_database.execute(s, {'a': fltr.tag_domain, 'b': fltr.tag_name}).fetchone()
+    assert r[0] == tx.hash

apps/cic-cache/tests/filters/test_faucet.py

@@ -0,0 +1,71 @@
+# standard imports
+import logging
+
+# external imports
+from chainlib.chain import ChainSpec
+from chainlib.eth.nonce import RPCNonceOracle
+from chainlib.eth.block import (
+        block_by_hash,
+        Block,
+        )
+from chainlib.eth.tx import (
+        receipt,
+        unpack,
+        transaction,
+        Tx,
+        )
+from hexathon import strip_0x
+from erc20_faucet.faucet import SingleShotFaucet
+from sqlalchemy import text
+
+# local imports
+from cic_cache.db import add_tag
+from cic_cache.runnable.daemons.filters.faucet import FaucetFilter
+
+logg = logging.getLogger()
+
+
+def test_filter_faucet(
+        eth_rpc,
+        eth_signer,
+        foo_token,
+        faucet_noregistry,
+        init_database,
+        list_defaults,
+        contract_roles,
+        agent_roles,
+        tags,
+        ):
+
+    chain_spec = ChainSpec('foo', 'bar', 42, 'baz')
+
+    fltr = FaucetFilter(chain_spec, contract_roles['CONTRACT_DEPLOYER'])
+
+    add_tag(init_database, fltr.tag_name, domain=fltr.tag_domain)
+
+    nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], eth_rpc)
+    c = SingleShotFaucet(chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle)
+    (tx_hash_hex, o) = c.give_to(faucet_noregistry, agent_roles['ALICE'], agent_roles['ALICE'])
+    r = eth_rpc.do(o)
+
+    tx_src = unpack(bytes.fromhex(strip_0x(o['params'][0])), chain_spec)
+
+    o = receipt(r)
+    r = eth_rpc.do(o)
+    rcpt = Tx.src_normalize(r)
+
+    assert r['status'] == 1
+
+    o = block_by_hash(r['block_hash'])
+    r = eth_rpc.do(o)
+    block_object = Block(r)
+
+    tx = Tx(tx_src, block_object)
+    tx.apply_receipt(rcpt)
+
+    r = fltr.filter(eth_rpc, block_object, tx, init_database)
+    assert r
+
+    s = text("SELECT x.tx_hash FROM tag a INNER JOIN tag_tx_link l ON l.tag_id = a.id INNER JOIN tx x ON x.id = l.tx_id WHERE a.domain = :a AND a.value = :b")
+    r = init_database.execute(s, {'a': fltr.tag_domain, 'b': fltr.tag_name}).fetchone()
+    assert r[0] == tx.hash

apps/cic-cache/tests/fixtures_config.py

@@ -2,7 +2,7 @@
 import os
 import logging
 
-# third-party imports
+# external imports
 import pytest
 import confini
 
@@ -13,7 +13,7 @@ logg = logging.getLogger(__file__)
 
 @pytest.fixture(scope='session')
 def load_config():
-    config_dir = os.path.join(root_dir, '.config/test')
+    config_dir = os.path.join(root_dir, 'config/test')
     conf = confini.Config(config_dir, 'CICTEST')
     conf.process()
     logg.debug('config {}'.format(conf))

apps/cic-cache/tests/fixtures_database.py

@@ -3,13 +3,16 @@ import os
 import logging
 import re
 
-# third-party imports
+# external imports
 import pytest
 import sqlparse
+import alembic
+from alembic.config import Config as AlembicConfig
 
 # local imports
 from cic_cache.db.models.base import SessionBase
 from cic_cache.db import dsn_from_config
+from cic_cache.db import add_tag
 
 logg = logging.getLogger(__file__)
 
@@ -26,11 +29,10 @@ def database_engine(
         except FileNotFoundError:
             pass
     dsn = dsn_from_config(load_config)
-    SessionBase.connect(dsn)
+    SessionBase.connect(dsn, debug=load_config.true('DATABASE_DEBUG'))
     return dsn
 
 
-# TODO: use alembic instead to migrate db, here we have to keep separate schema than migration script in script/migrate.py
 @pytest.fixture(scope='function')
 def init_database(
         load_config,
@@ -38,52 +40,23 @@ def init_database(
         ):
 
     rootdir = os.path.dirname(os.path.dirname(__file__))
-    schemadir = os.path.join(rootdir, 'db', load_config.get('DATABASE_DRIVER'))
-
-    if load_config.get('DATABASE_ENGINE') == 'sqlite':
-        rconn = SessionBase.engine.raw_connection()
-        f = open(os.path.join(schemadir, 'db.sql'))
-        s = f.read()
-        f.close()
-        rconn.executescript(s)
-
-    else:
-        rconn = SessionBase.engine.raw_connection()
-        rcursor = rconn.cursor()
-
-        #rcursor.execute('DROP FUNCTION IF EXISTS public.transaction_list')
-        #rcursor.execute('DROP FUNCTION IF EXISTS public.balances')
-
-        f = open(os.path.join(schemadir, 'db.sql'))
-        s = f.read()
-        f.close()
-        r = re.compile(r'^[A-Z]', re.MULTILINE)
-        for l in sqlparse.parse(s):
-            strl = str(l)
-            # we need to check for empty query lines, as sqlparse doesn't do that on its own (and psycopg complains when it gets them)
-            if not re.search(r, strl):
-                logg.warning('skipping parsed query line {}'.format(strl))
-                continue
-            rcursor.execute(strl)
-        rconn.commit()
-
-        rcursor.execute('SET search_path TO public')
-
-# this doesn't work when run separately, no idea why
-# functions have been manually added to original schema from cic-eth 
-#        f = open(os.path.join(schemadir, 'proc_transaction_list.sql'))
-#        s = f.read()
-#        f.close()
-#        rcursor.execute(s)
-# 
-#        f = open(os.path.join(schemadir, 'proc_balances.sql'))
-#        s = f.read()
-#        f.close()
-#        rcursor.execute(s)
-
-        rcursor.close()
+    dbdir = os.path.join(rootdir, 'cic_cache', 'db')
+    migrationsdir = os.path.join(dbdir, 'migrations', load_config.get('DATABASE_ENGINE'))
+    if not os.path.isdir(migrationsdir):
+        migrationsdir = os.path.join(dbdir, 'migrations', 'default')
+    logg.info('using migrations directory {}'.format(migrationsdir))
 
     session = SessionBase.create_session()
+
+    ac = AlembicConfig(os.path.join(migrationsdir, 'alembic.ini'))
+    ac.set_main_option('sqlalchemy.url', database_engine)
+    ac.set_main_option('script_location', migrationsdir)
+
+    alembic.command.downgrade(ac, 'base')
+    alembic.command.upgrade(ac, 'head')
+
+    session.commit()
+
     yield session
     session.commit()
     session.close()
@@ -116,3 +89,14 @@ def list_defaults(
     return {
         'block': 420000,
     }
+
+
+@pytest.fixture(scope='function')
+def tags(
+        init_database,
+        ):
+
+    add_tag(init_database, 'foo') 
+    add_tag(init_database, 'baz', domain='bar') 
+    add_tag(init_database, 'xyzzy', domain='bar') 
+    init_database.commit()

apps/cic-cache/tests/test_api.py

@@ -0,0 +1,31 @@
+# standard imports
+import json
+
+# external imports
+import pytest
+
+# local imports
+from cic_cache.runnable.daemons.query import process_transactions_all_data
+
+
+def test_api_all_data(
+        init_database,
+        txs,
+        ):
+
+    env = {
+        'PATH_INFO': '/txa/410000/420000',
+        'HTTP_X_CIC_CACHE_MODE': 'all',
+            }
+    j = process_transactions_all_data(init_database, env)
+    o = json.loads(j[1])
+    
+    assert len(o['data']) == 2
+
+    env = {
+        'PATH_INFO': '/txa/420000/410000',
+        'HTTP_X_CIC_CACHE_MODE': 'all',
+            }
+   
+    with pytest.raises(ValueError):
+        j = process_transactions_all_data(init_database, env)

apps/cic-cache/tests/test_cache.py

@@ -4,11 +4,12 @@ import datetime
 import logging
 import json
 
-# third-party imports
+# external imports
 import pytest
 
 # local imports
 from cic_cache import BloomCache
+from cic_cache.cache import DataCache
 
 logg = logging.getLogger()
 
@@ -33,3 +34,23 @@ def test_cache(
 
     assert b[0] == list_defaults['block'] - 1
 
+
+def test_cache_data(
+        init_database,
+        list_defaults,
+        list_actors,
+        list_tokens,
+        txs,
+        tag_txs,
+        ):
+
+    session = init_database
+
+    c = DataCache(session)
+    b = c.load_transactions_with_data(410000, 420000)
+
+    assert len(b[2]) == 2
+    assert b[2][0]['tx_hash'] == txs[1]
+    assert b[2][1]['tx_type'] == 'unknown'
+    assert b[2][0]['tx_type'] == 'test.taag'
+    

apps/cic-cache/tests/test_tag.py

@@ -0,0 +1,37 @@
+import os
+import datetime
+import logging
+import json
+
+# external imports
+import pytest
+
+# local imports
+from cic_cache.db import tag_transaction
+
+logg = logging.getLogger()
+
+
+def test_cache(
+        init_database,
+        list_defaults,
+        list_actors,
+        list_tokens,
+        txs,
+        tags,
+        ):
+
+    tag_transaction(init_database, txs[0], 'foo')
+    tag_transaction(init_database, txs[0], 'baz', domain='bar')
+    tag_transaction(init_database, txs[1], 'xyzzy', domain='bar')
+
+    r = init_database.execute("SELECT x.tx_hash FROM tag a INNER JOIN tag_tx_link l ON l.tag_id = a.id INNER JOIN tx x ON x.id = l.tx_id WHERE a.value = 'foo'").fetchall()
+    assert r[0][0] == txs[0]
+
+
+    r = init_database.execute("SELECT x.tx_hash FROM tag a INNER JOIN tag_tx_link l ON l.tag_id = a.id INNER JOIN tx x ON x.id = l.tx_id WHERE a.domain = 'bar' AND a.value = 'baz'").fetchall()
+    assert r[0][0] == txs[0]
+
+    
+    r = init_database.execute("SELECT x.tx_hash FROM tag a INNER JOIN tag_tx_link l ON l.tag_id = a.id INNER JOIN tx x ON x.id = l.tx_id WHERE a.domain = 'bar' AND a.value = 'xyzzy'").fetchall()
+    assert r[0][0] == txs[1]

apps/cic-eth/.coveragerc

@@ -5,3 +5,5 @@ omit =
 	cic_eth/db/migrations/*
 	cic_eth/sync/head.py
 	cic_eth/sync/mempool.py
+	cic_eth/queue/state.py
+	*redis*.py

apps/cic-eth/.gitlab-ci.yml

@@ -5,18 +5,29 @@
 
 .cic_eth_changes_target:
     rules:
-        - changes:
-            - $CONTEXT/$APP_NAME/*
+    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
+      #changes:
+          #- $CONTEXT/$APP_NAME/**/*
+      when: always
 
 build-mr-cic-eth:
     extends:
-        - .cic_eth_changes_target
-        - .py_build_merge_request
         - .cic_eth_variables
+        - .cic_eth_changes_target
+        - .py_build_target_test
+
+test-mr-cic-eth:
+    extends:
+        - .cic_eth_variables
+        - .cic_eth_changes_target
+    stage: test
+    image: $CI_REGISTRY_IMAGE/$APP_NAME-test:latest
+    script:
+        - cd apps/$APP_NAME/
+        - pytest -x --cov=cic_eth --cov-fail-under=90 --cov-report term-missing tests
+    needs: ["build-mr-cic-eth"]
 
 build-push-cic-eth:
     extends:
         - .py_build_push
         - .cic_eth_variables
-        
-

apps/cic-eth/cic_eth/admin/ctrl.py

@@ -2,7 +2,7 @@
 import datetime
 import logging
 
-# third-party imports
+# external imports
 import celery
 from chainlib.eth.constant import ZERO_ADDRESS
 from chainlib.chain import ChainSpec
@@ -32,7 +32,9 @@ def lock(chained_input, chain_spec_dict, address=ZERO_ADDRESS, flags=LockEnum.AL
     :returns: New lock state for address
     :rtype: number
     """
-    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
+    chain_str = '::'
+    if chain_spec_dict != None:
+        chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.set(chain_str, flags, address=address, tx_hash=tx_hash)
     logg.debug('Locked {} for {}, flag now {}'.format(flags, address, r))
     return chained_input
@@ -51,7 +53,9 @@ def unlock(chained_input, chain_spec_dict, address=ZERO_ADDRESS, flags=LockEnum.
     :returns: New lock state for address
     :rtype: number
     """
-    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
+    chain_str = '::'
+    if chain_spec_dict != None:
+        chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.reset(chain_str, flags, address=address)
     logg.debug('Unlocked {} for {}, flag now {}'.format(flags, address, r))
     return chained_input
@@ -127,7 +131,9 @@ def unlock_queue(chained_input, chain_spec_dict, address=ZERO_ADDRESS):
 
 @celery_app.task(base=CriticalSQLAlchemyTask)
 def check_lock(chained_input, chain_spec_dict, lock_flags, address=None):
-    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
+    chain_str = '::'
+    if chain_spec_dict != None:
+        chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     session = SessionBase.create_session()
     r = Lock.check(chain_str, lock_flags, address=ZERO_ADDRESS, session=session)
     if address != None:
@@ -139,3 +145,9 @@ def check_lock(chained_input, chain_spec_dict, lock_flags, address=None):
     session.flush()
     session.close()
     return chained_input
+
+
+@celery_app.task()
+def shutdown(message):
+    logg.critical('shutdown called: {}'.format(message))
+    celery_app.control.shutdown() #broadcast('shutdown')

apps/cic-eth/cic_eth/admin/nonce.py

@@ -4,12 +4,21 @@ import logging
 # external imports
 import celery
 from chainlib.chain import ChainSpec
-from chainlib.eth.tx import unpack
+from chainlib.connection import RPCConnection
+from chainlib.eth.tx import (
+        unpack,
+        TxFactory,
+        )
+from chainlib.eth.gas import OverrideGasOracle
+from chainqueue.sql.query import get_tx
+from chainqueue.sql.state import set_cancel
+from chainqueue.db.models.otx import Otx
+from chainqueue.db.models.tx import TxCache
+from hexathon import strip_0x
+from potaahto.symbols import snake_and_camel
 
 # local imports
 from cic_eth.db.models.base import SessionBase
-from cic_eth.db.models.otx import Otx
-from cic_eth.db.models.tx import TxCache
 from cic_eth.db.models.nonce import Nonce
 from cic_eth.admin.ctrl import (
         lock_send,
@@ -17,21 +26,16 @@ from cic_eth.admin.ctrl import (
         lock_queue,
         unlock_queue,
         )
-from cic_eth.queue.tx import (
-        get_tx,
-        set_cancel,
-        )
-from cic_eth.queue.tx import create as queue_create
-from cic_eth.eth.gas import (
-        create_check_gas_task,
-    )
+from cic_eth.queue.tx import queue_create
+from cic_eth.eth.gas import create_check_gas_task
+from cic_eth.task import BaseTask
 
 celery_app = celery.current_app
 logg = logging.getLogger()
 
 
-@celery_app.task(bind=True)
-def shift_nonce(self, chain_str, tx_hash_orig_hex, delta=1):
+@celery_app.task(bind=True, base=BaseTask)
+def shift_nonce(self, chainspec_dict, tx_hash_orig_hex, delta=1):
     """Shift all transactions with nonces higher than the offset by the provided position delta.
 
     Transactions who are replaced by transactions that move nonces will be marked as OVERRIDDEN.
@@ -42,25 +46,29 @@ def shift_nonce(self, chain_str, tx_hash_orig_hex, delta=1):
     :type tx_hash_orig_hex: str, 0x-hex
     :param delta: Amount
     """
+    chain_spec = ChainSpec.from_dict(chainspec_dict)
+    rpc = RPCConnection.connect(chain_spec, 'default')
+    rpc_signer = RPCConnection.connect(chain_spec, 'signer')
     queue = None
     try:
         queue = self.request.delivery_info.get('routing_key')
     except AttributeError:
         pass
 
-    chain_spec = ChainSpec.from_chain_str(chain_str)
-    tx_brief = get_tx(tx_hash_orig_hex)
-    tx_raw = bytes.fromhex(tx_brief['signed_tx'][2:])
-    tx = unpack(tx_raw, chain_spec.chain_id())
+    session = BaseTask.session_func()
+    tx_brief = get_tx(chain_spec, tx_hash_orig_hex, session=session)
+    tx_raw = bytes.fromhex(strip_0x(tx_brief['signed_tx']))
+    tx = unpack(tx_raw, chain_spec)
     nonce = tx_brief['nonce']
     address = tx['from']
 
-    logg.debug('shifting nonce {} position(s) for address {}, offset {}'.format(delta, address, nonce))
+    logg.debug('shifting nonce {} position(s) for address {}, offset {}, hash {}'.format(delta, address, nonce, tx['hash']))
 
-    lock_queue(None, chain_str, address)
-    lock_send(None, chain_str, address)
+    lock_queue(None, chain_spec.asdict(), address=address)
+    lock_send(None, chain_spec.asdict(), address=address)
+
+    set_cancel(chain_spec, strip_0x(tx['hash']), manual=True, session=session)
 
-    session = SessionBase.create_session()
     q = session.query(Otx)
     q = q.join(TxCache)
     q = q.filter(TxCache.sender==address)
@@ -71,51 +79,59 @@ def shift_nonce(self, chain_str, tx_hash_orig_hex, delta=1):
     tx_hashes = []
     txs = []
     for otx in otxs:
-        tx_raw = bytes.fromhex(otx.signed_tx[2:])
-        tx_new = unpack(tx_raw, chain_spec.chain_id())
+        tx_raw = bytes.fromhex(strip_0x(otx.signed_tx))
+        tx_new = unpack(tx_raw, chain_spec)
+        tx_new = snake_and_camel(tx_new)
 
         tx_previous_hash_hex = tx_new['hash']
         tx_previous_nonce = tx_new['nonce']
 
-        del(tx_new['hash'])
-        del(tx_new['hash_unsigned'])
+        tx_new['gas_price'] += 1
+        tx_new['gasPrice'] = tx_new['gas_price']
         tx_new['nonce'] -= delta
 
-        (tx_hash_hex, tx_signed_raw_hex) = sign_tx(tx_new, chain_str)
+        logg.debug('tx_new {}'.format(tx_new))
+
+        del(tx_new['hash'])
+        del(tx_new['hash_unsigned'])
+        del(tx_new['hashUnsigned'])
+
+        gas_oracle = OverrideGasOracle(limit=tx_new['gas'], price=tx_new['gas_price'] + 1) # TODO: it should be possible to merely set this price here and if missing in the existing struct then fill it in (chainlib.eth.tx)
+        c = TxFactory(chain_spec, signer=rpc_signer, gas_oracle=gas_oracle)
+        (tx_hash_hex, tx_signed_raw_hex) = c.build_raw(tx_new)
         logg.debug('tx {} -> {} nonce {} -> {}'.format(tx_previous_hash_hex, tx_hash_hex, tx_previous_nonce, tx_new['nonce']))
 
         otx = Otx(
-            nonce=tx_new['nonce'],
-            address=tx_new['from'],
-            tx_hash=tx_hash_hex,
-            signed_tx=tx_signed_raw_hex,
-                )
+            tx_new['nonce'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            )
         session.add(otx)
-        session.commit()
 
         # TODO: cancel all first, then replace. Otherwise we risk two non-locked states for two different nonces.
-        set_cancel(tx_previous_hash_hex, True)
+        set_cancel(chain_spec, strip_0x(tx_previous_hash_hex), manual=True, session=session)
 
-        TxCache.clone(tx_previous_hash_hex, tx_hash_hex)
+        TxCache.clone(tx_previous_hash_hex, tx_hash_hex, session=session)
 
         tx_hashes.append(tx_hash_hex)
         txs.append(tx_signed_raw_hex)
+        session.commit()
 
     session.close()
 
-    s = create_check_gas_and_send_task(
+    s = create_check_gas_task(
          txs, 
-         chain_str,
+         chain_spec,
          tx_new['from'],
-         tx_new['gas'],
-         tx_hashes, 
-         queue,
+         gas=tx_new['gas'],
+         tx_hashes_hex=tx_hashes, 
+         queue=queue,
         )
 
     s_unlock_send = celery.signature(
         'cic_eth.admin.ctrl.unlock_send',
         [
-            chain_str,
+            chain_spec.asdict(),
             tx_new['from'],
             ],
         queue=queue,
@@ -123,7 +139,7 @@ def shift_nonce(self, chain_str, tx_hash_orig_hex, delta=1):
     s_unlock_direct = celery.signature(
         'cic_eth.admin.ctrl.unlock_queue',
         [
-            chain_str,
+            chain_spec.asdict(),
             tx_new['from'],
             ],
         queue=queue,

apps/cic-eth/cic_eth/admin/token.py

@@ -0,0 +1,21 @@
+# standard imports
+import logging
+
+# external imports
+import celery
+
+# local imports
+from cic_eth.task import BaseTask
+
+celery_app = celery.current_app
+logg = logging.getLogger()
+
+
+@celery_app.task(bind=True, base=BaseTask)
+def default_token(self):
+    return {
+            'symbol': self.default_token_symbol,
+            'address': self.default_token_address,
+            'name': self.default_token_name,
+            'decimals': self.default_token_decimals,
+            }

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

@@ -8,6 +8,7 @@ from chainlib.eth.constant import (
         ZERO_ADDRESS,
         )
 from cic_eth_registry import CICRegistry
+from cic_eth_registry.erc20 import ERC20Token
 from cic_eth_registry.error import UnknownContractError
 from chainlib.eth.address import to_checksum_address
 from chainlib.eth.contract import code
@@ -22,23 +23,22 @@ from hexathon import (
         add_0x,
         )
 from chainlib.eth.gas import balance
-
-# local imports
-from cic_eth.db.models.base import SessionBase
-from cic_eth.db.models.role import AccountRole
-from cic_eth.db.models.otx import Otx
-from cic_eth.db.models.tx import TxCache
-from cic_eth.db.models.nonce import Nonce
-from cic_eth.db.enum import (
+from chainqueue.db.enum import (
         StatusEnum,
         StatusBits,
         is_alive,
         is_error_status,
         status_str,
     )
+from chainqueue.error import TxStateChangeError
+from chainqueue.query import get_tx
+from eth_erc20 import ERC20
+
+# local imports
+from cic_eth.db.models.base import SessionBase
+from cic_eth.db.models.role import AccountRole
+from cic_eth.db.models.nonce import Nonce
 from cic_eth.error import InitializationError
-from cic_eth.db.error import TxStateChangeError
-from cic_eth.queue.tx import get_tx
 
 app = celery.current_app
 
@@ -62,6 +62,29 @@ class AdminApi:
         self.call_address = call_address
 
 
+    def proxy_do(self, chain_spec, o):
+        s_proxy = celery.signature(
+                'cic_eth.task.rpc_proxy',
+                [
+                    chain_spec.asdict(),
+                    o,
+                    'default',
+                    ],
+                queue=self.queue
+                )
+        return s_proxy.apply_async()
+
+
+    
+    def registry(self):
+        s_registry = celery.signature(
+                'cic_eth.task.registry',
+                [],
+                queue=self.queue
+                )
+        return s_registry.apply_async()
+
+
     def unlock(self, chain_spec, address, flags=None):
         s_unlock = celery.signature(
             'cic_eth.admin.ctrl.unlock',
@@ -92,7 +115,7 @@ class AdminApi:
 
     def get_lock(self):
         s_lock = celery.signature(
-            'cic_eth.queue.tx.get_lock',
+            'cic_eth.queue.lock.get_lock',
             [],
             queue=self.queue,
             )
@@ -138,8 +161,9 @@ class AdminApi:
 
         logg.debug('resend {}'.format(tx_hash_hex))
         s_get_tx_cache = celery.signature(
-            'cic_eth.queue.tx.get_tx_cache',
+            'cic_eth.queue.query.get_tx_cache',
             [
+                chain_spec.asdict(),
                 tx_hash_hex,
                 ],
             queue=self.queue,
@@ -147,7 +171,6 @@ class AdminApi:
 
         # TODO: This check should most likely be in resend task itself
         tx_dict = s_get_tx_cache.apply_async().get()
-        #if tx_dict['status'] in [StatusEnum.REVERTED, StatusEnum.SUCCESS, StatusEnum.CANCELLED, StatusEnum.OBSOLETED]: 
         if not is_alive(getattr(StatusEnum, tx_dict['status']).value):
             raise TxStateChangeError('Cannot resend mined or obsoleted transaction'.format(txold_hash_hex))
         
@@ -155,7 +178,7 @@ class AdminApi:
             raise NotImplementedError('resend as new not yet implemented')
 
         s = celery.signature(
-            'cic_eth.eth.tx.resend_with_higher_gas',
+            'cic_eth.eth.gas.resend_with_higher_gas',
             [
                 chain_spec.asdict(),
                 None,
@@ -165,8 +188,9 @@ class AdminApi:
             )
 
         s_manual = celery.signature(
-            'cic_eth.queue.tx.set_manual',
+            'cic_eth.queue.state.set_manual',
             [
+                chain_spec.asdict(),
                 tx_hash_hex,
                 ],
             queue=self.queue,
@@ -185,11 +209,13 @@ class AdminApi:
             s.link(s_gas)
 
         return s_manual.apply_async()
-                        
-    def check_nonce(self, address):
+                    
+
+    def check_nonce(self, chain_spec, address):
         s = celery.signature(
-                'cic_eth.queue.tx.get_account_tx',
+                'cic_eth.queue.query.get_account_tx',
                 [
+                    chain_spec.asdict(),
                     address,
                     True,
                     False,
@@ -204,14 +230,14 @@ class AdminApi:
         last_nonce = -1
         for k in txs.keys():
             s_get_tx = celery.signature(
-                    'cic_eth.queue.tx.get_tx',
+                    'cic_eth.queue.query.get_tx',
                     [
+                        chain_spec.asdict(),
                         k,
                         ],
                     queue=self.queue,
                     )
             tx = s_get_tx.apply_async().get()
-            #tx = get_tx(k)
             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:
@@ -219,15 +245,14 @@ class AdminApi:
                 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']))
+                logg.debug('tx {}'.format(tx))
+                tx_obj = unpack(bytes.fromhex(strip_0x(tx['signed_tx'])), chain_spec)
+                logg.error('nonce gap; {} followed {} for account {}'.format(nonce_otx, last_nonce, tx_obj['from']))
                 blocking_tx = k
                 blocking_nonce = nonce_otx
                 break
             last_nonce = nonce_otx
 
-        #nonce_cache = Nonce.get(address)
-        #nonce_w3 = self.w3.eth.getTransactionCount(address, 'pending') 
-        
         return {
             'nonce': {
                 #'network': nonce_cache,
@@ -236,15 +261,17 @@ class AdminApi:
                 'blocking': blocking_nonce,
             },
             'tx': {
-                'blocking': blocking_tx,
-                }
+                'blocking': add_0x(blocking_tx),
             }
+        }
 
 
-    def fix_nonce(self, address, nonce, chain_spec):
+    # TODO: is risky since it does not validate that there is actually a nonce problem?
+    def fix_nonce(self, chain_spec, address, nonce):
         s = celery.signature(
-                'cic_eth.queue.tx.get_account_tx',
+                'cic_eth.queue.query.get_account_tx',
                 [
+                    chain_spec.asdict(),
                     address,
                     True,
                     False,
@@ -254,15 +281,17 @@ class AdminApi:
         txs = s.apply_async().get()
 
         tx_hash_hex = None
+        session = SessionBase.create_session()
         for k in txs.keys():
-            tx_dict = get_tx(k)
+            tx_dict = get_tx(chain_spec, k, session=session)
             if tx_dict['nonce'] == nonce:
                 tx_hash_hex = k
+        session.close()
 
         s_nonce = celery.signature(
                 'cic_eth.admin.nonce.shift_nonce',
                 [
-                    self.rpc.chain_spec.asdict(),
+                    chain_spec.asdict(),
                     tx_hash_hex, 
                 ],
                 queue=self.queue
@@ -270,20 +299,6 @@ class AdminApi:
         return s_nonce.apply_async()
 
 
-#    # TODO: this is a stub, complete all checks
-#    def ready(self):
-#        """Checks whether all required initializations have been performed.
-#
-#        :raises cic_eth.error.InitializationError: At least one setting pre-requisite has not been met.
-#        :raises KeyError: An address provided for initialization is not known by the keystore.
-#        """
-#        addr = AccountRole.get_address('ETH_GAS_PROVIDER_ADDRESS')
-#        if addr == ZERO_ADDRESS:
-#            raise InitializationError('missing account ETH_GAS_PROVIDER_ADDRESS')
-#
-#        self.w3.eth.sign(addr, text='666f6f')
-
-
     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.
 
@@ -294,8 +309,9 @@ class AdminApi:
         """
         last_nonce = -1
         s = celery.signature(
-                'cic_eth.queue.tx.get_account_tx',
+                'cic_eth.queue.query.get_account_tx',
                 [
+                    chain_spec.asdict(),
                     address,
                     ],
                 queue=self.queue,
@@ -306,8 +322,11 @@ class AdminApi:
         for tx_hash in txs.keys():
             errors = []
             s = celery.signature(
-                    'cic_eth.queue.tx.get_tx_cache',
-                    [tx_hash],
+                    'cic_eth.queue.query.get_tx_cache',
+                    [
+                        chain_spec.asdict(),
+                        tx_hash,
+                        ],
                     queue=self.queue,
                     )
             tx_dict = s.apply_async().get()
@@ -342,6 +361,7 @@ class AdminApi:
 
 
     # TODO: Add exception upon non-existent tx aswell as invalid tx data to docstring 
+    # TODO: This method is WAY too long
     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.
 
@@ -364,50 +384,99 @@ class AdminApi:
 
         if tx_raw != None:
             tx_hash = add_0x(keccak256_hex_to_hex(tx_raw))
-            #tx_hash = self.w3.keccak(hexstr=tx_raw).hex()
 
         s = celery.signature(
-            'cic_eth.queue.tx.get_tx_cache',
-            [tx_hash],
+            'cic_eth.queue.query.get_tx_cache',
+            [
+                chain_spec.asdict(),
+                tx_hash,
+                ],
             queue=self.queue,
             )
     
-        tx = s.apply_async().get()
-  
+        t = s.apply_async()
+        tx = t.get()
+
         source_token = None
         if tx['source_token'] != ZERO_ADDRESS:
-            try:
-                source_token = registry.by_address(tx['source_token'])
-                #source_token = CICRegistry.get_address(chain_spec, tx['source_token']).contract
-            except UnknownContractError:
-                #source_token_contract = self.w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=tx['source_token'])
-                #source_token = CICRegistry.add_token(chain_spec, source_token_contract)
-                logg.warning('unknown source token contract {}'.format(tx['source_token']))
+            source_token_declaration = None
+            if registry != None:
+                try:
+                    source_token_declaration = registry.by_address(tx['source_token'], sender_address=self.call_address)
+                except UnknownContractError:
+                    logg.warning('unknown source token contract {} (direct)'.format(tx['source_token']))
+            else:
+                s = celery.signature(
+                        'cic_eth.task.registry_address_lookup',
+                        [
+                            chain_spec.asdict(),
+                            tx['source_token'],
+                            ],
+                        queue=self.queue
+                        )
+                t = s.apply_async()
+                source_token_declaration = t.get()
+
+            if source_token_declaration != None:
+                logg.warning('found declarator record for source token {} but not checking validity'.format(tx['source_token']))
+                source_token = ERC20Token(chain_spec, self.rpc, tx['source_token'])
+                logg.debug('source token set tup {}'.format(source_token))
+
+
 
         destination_token = None
-        if tx['source_token'] != ZERO_ADDRESS:
-            try:
-                #destination_token = CICRegistry.get_address(chain_spec, tx['destination_token'])
-                destination_token = registry.by_address(tx['destination_token'])
-            except UnknownContractError:
-                #destination_token_contract = self.w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=tx['source_token'])
-                #destination_token = CICRegistry.add_token(chain_spec, destination_token_contract)
-                logg.warning('unknown destination token contract {}'.format(tx['destination_token']))
+        if tx['destination_token'] != ZERO_ADDRESS:
+            destination_token_declaration = None
+            if registry != None:
+                try:
+                    destination_token_declaration = registry.by_address(tx['destination_token'], sender_address=self.call_address)
+                except UnknownContractError:
+                    logg.warning('unknown destination token contract {}'.format(tx['destination_token']))
+            else:
+                s = celery.signature(
+                        'cic_eth.task.registry_address_lookup',
+                        [
+                            chain_spec.asdict(),
+                            tx['destination_token'],
+                            ],
+                        queue=self.queue
+                        )
+                t = s.apply_async()
+                destination_token_declaration = t.get()
+            if destination_token_declaration != None:
+                logg.warning('found declarator record for destination token {} but not checking validity'.format(tx['destination_token']))
+                destination_token = ERC20Token(chain_spec, self.rpc, tx['destination_token'])
 
         tx['sender_description'] = 'Custodial account'
         tx['recipient_description'] = 'Custodial account'
 
         o = code(tx['sender'])
-        r = self.rpc.do(o)
+        t = self.proxy_do(chain_spec, o)
+        r = t.get()
         if len(strip_0x(r, allow_empty=True)) > 0:
-            try:
-                #sender_contract = CICRegistry.get_address(chain_spec, tx['sender'])
-                sender_contract = registry.by_address(tx['sender'], sender_address=self.call_address)
-                tx['sender_description'] = 'Contract at {}'.format(tx['sender']) #sender_contract)
-            except UnknownContractError:
-                tx['sender_description'] = 'Unknown contract'
-            except KeyError as e:
-                tx['sender_description'] = 'Unknown contract'
+            if registry != None:
+                try:
+                    sender_contract = registry.by_address(tx['sender'], sender_address=self.call_address)
+                    tx['sender_description'] = 'Contract at {}'.format(tx['sender'])
+                except UnknownContractError:
+                    tx['sender_description'] = 'Unknown contract'
+                except KeyError as e:
+                    tx['sender_description'] = 'Unknown contract'
+            else:
+                s = celery.signature(
+                        'cic_eth.task.registry_address_lookup',
+                        [
+                            chain_spec.asdict(),
+                            tx['sender'],
+                            ],
+                        queue=self.queue
+                        )
+                t = s.apply_async()
+                tx['sender_description'] = t.get()
+                if tx['sender_description'] == None:
+                    tx['sender_description'] = 'Unknown contract'
+
+
         else:
             s = celery.signature(
                     'cic_eth.eth.account.have',
@@ -436,16 +505,31 @@ class AdminApi:
                     tx['sender_description'] = role
 
         o = code(tx['recipient'])
-        r = self.rpc.do(o)
+        t = self.proxy_do(chain_spec, o)
+        r = t.get()
         if len(strip_0x(r, allow_empty=True)) > 0:
-            try:
-                #recipient_contract = CICRegistry.by_address(tx['recipient'])
-                recipient_contract = registry.by_address(tx['recipient'])
-                tx['recipient_description'] = 'Contract at {}'.format(tx['recipient']) #recipient_contract)
-            except UnknownContractError as e:
-                tx['recipient_description'] = 'Unknown contract'
-            except KeyError as e:
-                tx['recipient_description'] = 'Unknown contract'
+            if registry != None:
+                try:
+                    recipient_contract = registry.by_address(tx['recipient'])
+                    tx['recipient_description'] = 'Contract at {}'.format(tx['recipient'])
+                except UnknownContractError as e:
+                    tx['recipient_description'] = 'Unknown contract'
+                except KeyError as e:
+                    tx['recipient_description'] = 'Unknown contract'
+            else:
+                s = celery.signature(
+                        'cic_eth.task.registry_address_lookup',
+                        [
+                            chain_spec.asdict(),
+                            tx['recipient'],
+                            ],
+                        queue=self.queue
+                        )
+                t = s.apply_async()
+                tx['recipient_description'] = t.get()
+                if tx['recipient_description'] == None:
+                    tx['recipient_description'] = 'Unknown contract'
+
         else:
             s = celery.signature(
                     'cic_eth.eth.account.have',
@@ -473,13 +557,19 @@ class AdminApi:
                 if role != None:
                     tx['recipient_description'] = role
 
+        erc20_c = ERC20(chain_spec)
         if source_token != None:
-            tx['source_token_symbol'] = source_token.symbol()
-            tx['sender_token_balance'] = source_token.function('balanceOf')(tx['sender']).call()
+            tx['source_token_symbol'] = source_token.symbol
+            o = erc20_c.balance_of(tx['source_token'], tx['sender'], sender_address=self.call_address)
+            r = self.rpc.do(o)
+            tx['sender_token_balance'] = erc20_c.parse_balance_of(r)
 
         if destination_token != None:
-            tx['destination_token_symbol'] = destination_token.symbol()
-            tx['recipient_token_balance'] = source_token.function('balanceOf')(tx['recipient']).call()
+            tx['destination_token_symbol'] = destination_token.symbol
+            o = erc20_c.balance_of(tx['destination_token'], tx['recipient'], sender_address=self.call_address)
+            r = self.rpc.do(o)
+            tx['recipient_token_balance'] = erc20_c.parse_balance_of(r)
+            #tx['recipient_token_balance'] = destination_token.function('balanceOf')(tx['recipient']).call()
 
         # 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' 
@@ -487,7 +577,8 @@ class AdminApi:
         r = None
         try:
             o = transaction(tx_hash)
-            r = self.rpc.do(o)
+            t = self.proxy_do(chain_spec, o)
+            r = t.get()
             if r != None:
                 tx['network_status'] = 'Mempool'
         except Exception as e:
@@ -496,7 +587,8 @@ class AdminApi:
         if r != None:
             try:
                 o = receipt(tx_hash)
-                r = self.rpc.do(o)
+                t = self.proxy_do(chain_spec, o)
+                r = t.get()
                 logg.debug('h {} o {}'.format(tx_hash, o))
                 if int(strip_0x(r['status'])) == 1:
                     tx['network_status'] = 'Confirmed'
@@ -511,21 +603,24 @@ class AdminApi:
                 pass
 
         o = balance(tx['sender'])
-        r = self.rpc.do(o)
+        t = self.proxy_do(chain_spec, o)
+        r = t.get()
         tx['sender_gas_balance'] = r
 
         o = balance(tx['recipient'])
-        r = self.rpc.do(o)
+        t = self.proxy_do(chain_spec, o)
+        r = t.get()
         tx['recipient_gas_balance'] = r
 
-        tx_unpacked = unpack(bytes.fromhex(tx['signed_tx'][2:]), chain_spec.chain_id())
+        tx_unpacked = unpack(bytes.fromhex(strip_0x(tx['signed_tx'])), chain_spec)
         tx['gas_price'] = tx_unpacked['gasPrice']
         tx['gas_limit'] = tx_unpacked['gas']
         tx['data'] = tx_unpacked['data']
 
         s = celery.signature(
-            'cic_eth.queue.tx.get_state_log',
+            'cic_eth.queue.state.get_state_log',
             [
+                chain_spec.asdict(),
                 tx_hash,
                 ],
             queue=self.queue,

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

@@ -37,7 +37,7 @@ class Api:
         self.callback_param = callback_param
         self.callback_task = callback_task
         self.queue = queue
-        logg.info('api using queue {}'.format(self.queue))
+        logg.debug('api using queue {}'.format(self.queue))
         self.callback_success = None
         self.callback_error = None
         if callback_queue == None:
@@ -62,130 +62,146 @@ class Api:
                     )       
 
 
-    def convert_transfer(self, from_address, to_address, target_return, minimum_return, from_token_symbol, to_token_symbol):
-        """Executes a chain of celery tasks that performs conversion between two ERC20 tokens, and transfers to a specified receipient after convert has completed.
-
-        :param from_address: Ethereum address of sender
-        :type from_address: str, 0x-hex
-        :param to_address: Ethereum address of receipient
-        :type to_address: str, 0x-hex
-        :param target_return: Estimated return from conversion
-        :type  target_return: int
-        :param minimum_return: The least value of destination token return to allow
-        :type minimum_return: int
-        :param from_token_symbol: ERC20 token symbol of token being converted
-        :type from_token_symbol: str
-        :param to_token_symbol: ERC20 token symbol of token to receive
-        :type to_token_symbol: str
-        :returns: uuid of root task
-        :rtype: celery.Task
-        """
-        raise NotImplementedError('out of service until new DEX migration is done')
-        s_check = celery.signature(
-                'cic_eth.admin.ctrl.check_lock',
-                [
-                    [from_token_symbol, to_token_symbol],
-                    self.chain_spec.asdict(),
-                    LockEnum.QUEUE,
-                    from_address,
-                    ],
-                queue=self.queue,
-                )
-        s_nonce = celery.signature(
-                'cic_eth.eth.tx.reserve_nonce',
+    def default_token(self):
+        s_token = celery.signature(
+                'cic_eth.admin.token.default_token',
                 [],
                 queue=self.queue,
                 )
-        s_tokens = celery.signature(
-                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
-                [
-                    self.chain_str,
-                    ],
-                queue=self.queue,
-                )
-        s_convert = celery.signature(
-                'cic_eth.eth.bancor.convert_with_default_reserve',
-                [
-                    from_address,
-                    target_return,
-                    minimum_return,
-                    to_address,
-                    self.chain_spec.asdict(),
-                    ],
-                queue=self.queue,
-                )
-        s_nonce.link(s_tokens)
-        s_check.link(s_nonce)
         if self.callback_param != None:
-            s_convert.link(self.callback_success)
-            s_tokens.link(s_convert).on_error(self.callback_error)
-        else:
-            s_tokens.link(s_convert)
+            s_token.link(self.callback_success)
 
-        t = s_check.apply_async(queue=self.queue)
-        return t
+        return s_token.apply_async()
 
 
-    def convert(self, from_address, target_return, minimum_return, from_token_symbol, to_token_symbol):
-        """Executes a chain of celery tasks that performs conversion between two ERC20 tokens.
-
-        :param from_address: Ethereum address of sender
-        :type from_address: str, 0x-hex
-        :param target_return: Estimated return from conversion
-        :type  target_return: int
-        :param minimum_return: The least value of destination token return to allow
-        :type minimum_return: int
-        :param from_token_symbol: ERC20 token symbol of token being converted
-        :type from_token_symbol: str
-        :param to_token_symbol: ERC20 token symbol of token to receive
-        :type to_token_symbol: str
-        :returns: uuid of root task
-        :rtype: celery.Task
-        """
-        raise NotImplementedError('out of service until new DEX migration is done')
-        s_check = celery.signature(
-                'cic_eth.admin.ctrl.check_lock',
-                [
-                    [from_token_symbol, to_token_symbol],
-                    self.chain_spec.asdict(),
-                    LockEnum.QUEUE,
-                    from_address,
-                    ],
-                queue=self.queue,
-                )
-        s_nonce = celery.signature(
-                'cic_eth.eth.tx.reserve_nonce',
-                [],
-                queue=self.queue,
-                )
-        s_tokens = celery.signature(
-                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
-                [
-                    self.chain_spec.asdict(),
-                    ],
-                queue=self.queue,
-                )
-        s_convert = celery.signature(
-                'cic_eth.eth.bancor.convert_with_default_reserve',
-                [
-                    from_address,
-                    target_return,
-                    minimum_return,
-                    from_address,
-                    self.chain_spec.asdict(),
-                    ],
-                queue=self.queue,
-                )
-        s_nonce.link(s_tokens)
-        s_check.link(s_nonce)
-        if self.callback_param != None:
-            s_convert.link(self.callback_success)
-            s_tokens.link(s_convert).on_error(self.callback_error)
-        else:
-            s_tokens.link(s_convert)
-
-        t = s_check.apply_async(queue=self.queue)
-        return t
+#    def convert_transfer(self, from_address, to_address, target_return, minimum_return, from_token_symbol, to_token_symbol):
+#        """Executes a chain of celery tasks that performs conversion between two ERC20 tokens, and transfers to a specified receipient after convert has completed.
+#
+#        :param from_address: Ethereum address of sender
+#        :type from_address: str, 0x-hex
+#        :param to_address: Ethereum address of receipient
+#        :type to_address: str, 0x-hex
+#        :param target_return: Estimated return from conversion
+#        :type  target_return: int
+#        :param minimum_return: The least value of destination token return to allow
+#        :type minimum_return: int
+#        :param from_token_symbol: ERC20 token symbol of token being converted
+#        :type from_token_symbol: str
+#        :param to_token_symbol: ERC20 token symbol of token to receive
+#        :type to_token_symbol: str
+#        :returns: uuid of root task
+#        :rtype: celery.Task
+#        """
+#        raise NotImplementedError('out of service until new DEX migration is done')
+#        s_check = celery.signature(
+#                'cic_eth.admin.ctrl.check_lock',
+#                [
+#                    [from_token_symbol, to_token_symbol],
+#                    self.chain_spec.asdict(),
+#                    LockEnum.QUEUE,
+#                    from_address,
+#                    ],
+#                queue=self.queue,
+#                )
+#        s_nonce = celery.signature(
+#                'cic_eth.eth.nonce.reserve_nonce',
+#                [
+#                    self.chain_spec.asdict(),
+#                    ],
+#                queue=self.queue,
+#                )
+#        s_tokens = celery.signature(
+#                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
+#                [
+#                    self.chain_str,
+#                    ],
+#                queue=self.queue,
+#                )
+#        s_convert = celery.signature(
+#                'cic_eth.eth.bancor.convert_with_default_reserve',
+#                [
+#                    from_address,
+#                    target_return,
+#                    minimum_return,
+#                    to_address,
+#                    self.chain_spec.asdict(),
+#                    ],
+#                queue=self.queue,
+#                )
+#        s_nonce.link(s_tokens)
+#        s_check.link(s_nonce)
+#        if self.callback_param != None:
+#            s_convert.link(self.callback_success)
+#            s_tokens.link(s_convert).on_error(self.callback_error)
+#        else:
+#            s_tokens.link(s_convert)
+#
+#        t = s_check.apply_async(queue=self.queue)
+#        return t
+#
+#
+#    def convert(self, from_address, target_return, minimum_return, from_token_symbol, to_token_symbol):
+#        """Executes a chain of celery tasks that performs conversion between two ERC20 tokens.
+#
+#        :param from_address: Ethereum address of sender
+#        :type from_address: str, 0x-hex
+#        :param target_return: Estimated return from conversion
+#        :type  target_return: int
+#        :param minimum_return: The least value of destination token return to allow
+#        :type minimum_return: int
+#        :param from_token_symbol: ERC20 token symbol of token being converted
+#        :type from_token_symbol: str
+#        :param to_token_symbol: ERC20 token symbol of token to receive
+#        :type to_token_symbol: str
+#        :returns: uuid of root task
+#        :rtype: celery.Task
+#        """
+#        raise NotImplementedError('out of service until new DEX migration is done')
+#        s_check = celery.signature(
+#                'cic_eth.admin.ctrl.check_lock',
+#                [
+#                    [from_token_symbol, to_token_symbol],
+#                    self.chain_spec.asdict(),
+#                    LockEnum.QUEUE,
+#                    from_address,
+#                    ],
+#                queue=self.queue,
+#                )
+#        s_nonce = celery.signature(
+#                'cic_eth.eth.nonce.reserve_nonce',
+#                [
+#                    self.chain_spec.asdict(),
+#                    ],
+#                queue=self.queue,
+#                )
+#        s_tokens = celery.signature(
+#                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
+#                [
+#                    self.chain_spec.asdict(),
+#                    ],
+#                queue=self.queue,
+#                )
+#        s_convert = celery.signature(
+#                'cic_eth.eth.bancor.convert_with_default_reserve',
+#                [
+#                    from_address,
+#                    target_return,
+#                    minimum_return,
+#                    from_address,
+#                    self.chain_spec.asdict(),
+#                    ],
+#                queue=self.queue,
+#                )
+#        s_nonce.link(s_tokens)
+#        s_check.link(s_nonce)
+#        if self.callback_param != None:
+#            s_convert.link(self.callback_success)
+#            s_tokens.link(s_convert).on_error(self.callback_error)
+#        else:
+#            s_tokens.link(s_convert)
+#
+#        t = s_check.apply_async(queue=self.queue)
+#        return t
 
 
     def transfer(self, from_address, to_address, value, token_symbol):
@@ -213,8 +229,9 @@ class Api:
                 queue=self.queue,
                 )
         s_nonce = celery.signature(
-                'cic_eth.eth.tx.reserve_nonce',
+                'cic_eth.eth.nonce.reserve_nonce',
                 [
+                    self.chain_spec.asdict(),
                     from_address,
                     ],
                 queue=self.queue,
@@ -359,8 +376,9 @@ class Api:
 
         if register:
             s_nonce = celery.signature(
-                'cic_eth.eth.tx.reserve_nonce',
+                'cic_eth.eth.nonce.reserve_nonce',
                 [
+                    self.chain_spec.asdict(),
                     'ACCOUNT_REGISTRY_WRITER',
                     ],
                 queue=self.queue,
@@ -397,14 +415,15 @@ class Api:
                 queue=self.queue,
                 )
         s_nonce = celery.signature(
-                'cic_eth.eth.tx.reserve_nonce',
+                'cic_eth.eth.nonce.reserve_nonce',
                 [
+                    self.chain_spec.asdict(),
                     'GAS_GIFTER',
                     ],
                 queue=self.queue,
                 )
         s_refill = celery.signature(
-                'cic_eth.eth.tx.refill_gas',
+                'cic_eth.eth.gas.refill_gas',
                 [
                     self.chain_spec.asdict(),
                     ],
@@ -439,8 +458,9 @@ class Api:
         """
         offset = 0
         s_local = celery.signature(
-            'cic_eth.queue.tx.get_account_tx',
+            'cic_eth.queue.query.get_account_tx',
             [
+                self.chain_spec.asdict(),
                 address,
                 ],
             queue=self.queue,

apps/cic-eth/cic_eth/check/db.py

@@ -0,0 +1,8 @@
+from cic_eth.db.models.base import SessionBase
+
+
+def health(*args, **kwargs):
+    session = SessionBase.create_session()
+    session.execute('SELECT count(*) from alembic_version')
+    session.close()
+    return True

apps/cic-eth/cic_eth/check/gas.py

@@ -0,0 +1,48 @@
+# standard imports
+import logging
+
+# external imports
+from chainlib.connection import RPCConnection
+from chainlib.chain import ChainSpec
+from chainlib.eth.gas import balance
+
+# local imports
+from cic_eth.db.models.role import AccountRole
+from cic_eth.db.models.base import SessionBase
+from cic_eth.db.enum import LockEnum
+from cic_eth.error import LockedError
+from cic_eth.admin.ctrl import check_lock
+
+logg = logging.getLogger().getChild(__name__)
+
+
+def health(*args, **kwargs):
+
+    session = SessionBase.create_session()
+
+    config = kwargs['config']
+    chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
+    logg.debug('check gas balance of gas gifter for chain {}'.format(chain_spec))
+
+    try:
+        check_lock(None, None, LockEnum.INIT)
+    except LockedError:
+        logg.warning('INIT lock is set, skipping GAS GIFTER balance check.')
+        return True
+
+    gas_provider = AccountRole.get_address('GAS_GIFTER', session=session)
+    session.close()
+
+    rpc = RPCConnection.connect(chain_spec, 'default')
+    o = balance(gas_provider)
+    r = rpc.do(o)
+    try: 
+        r = int(r, 16)
+    except TypeError:
+        r = int(r)
+    gas_min = int(config.get('ETH_GAS_GIFTER_MINIMUM_BALANCE'))
+    if r < gas_min:
+        logg.error('EEK! gas gifter has balance {}, below minimum {}'.format(r, gas_min))
+        return False
+
+    return True

apps/cic-eth/cic_eth/check/redis.py

@@ -0,0 +1,18 @@
+# external imports
+import redis
+import os
+
+
+def health(*args, **kwargs):
+    r = redis.Redis(
+            host=kwargs['config'].get('REDIS_HOST'),
+            port=kwargs['config'].get('REDIS_PORT'),
+            db=kwargs['config'].get('REDIS_DB'),
+            )
+    try:
+        r.set(kwargs['unit'], os.getpid())
+    except redis.connection.ConnectionError:
+        return False
+    except redis.connection.ResponseError:
+        return False
+    return True

apps/cic-eth/cic_eth/check/signer.py

@@ -0,0 +1,37 @@
+# standard imports
+import time
+import logging
+from urllib.error import URLError
+
+# external imports
+from chainlib.connection import RPCConnection
+from chainlib.eth.constant import ZERO_ADDRESS
+from chainlib.eth.sign import sign_message
+from chainlib.error import JSONRPCException
+
+logg = logging.getLogger().getChild(__name__)
+
+
+def health(*args, **kwargs):
+    blocked = True
+    max_attempts = 5
+    conn = RPCConnection.connect(kwargs['config'].get('CIC_CHAIN_SPEC'), tag='signer')
+    for i in range(max_attempts):
+        idx = i + 1
+        logg.debug('attempt signer connection check {}/{}'.format(idx, max_attempts))
+        try:
+            conn.do(sign_message(ZERO_ADDRESS, '0x2a'))
+        except FileNotFoundError:
+            pass
+        except ConnectionError:
+            pass
+        except URLError:
+            pass
+        except JSONRPCException:
+            logg.debug('signer connection succeeded')
+            return True
+
+        if idx < max_attempts:
+            time.sleep(0.5)
+
+    return False

apps/cic-eth/cic_eth/db/__init__.py

@@ -11,10 +11,6 @@ logg = logging.getLogger()
 # an Engine, which the Session will use for connection
 # resources
 
-# TODO: Remove the package exports, all models should be imported using full path
-from .models.otx import Otx
-from .models.convert import TxConvertTransfer
-
 
 def dsn_from_config(config):
     """Generate a dsn string from the provided config dict.

apps/cic-eth/cic_eth/db/enum.py

@@ -74,10 +74,11 @@ class LockEnum(enum.IntEnum):
     QUEUE: Disable queueing new or modified transactions
     """
     STICKY=1
-    CREATE=2
-    SEND=4
-    QUEUE=8
-    QUERY=16
+    INIT=2
+    CREATE=4
+    SEND=8
+    QUEUE=16
+    QUERY=32
     ALL=int(0xfffffffffffffffe)
 
 

apps/cic-eth/cic_eth/db/migrations/default/versions/0ec0d6d1e785_add_chainqueue.py

@@ -0,0 +1,29 @@
+"""Add chainqueue
+
+Revision ID: 0ec0d6d1e785
+Revises: 
+Create Date: 2021-04-02 18:30:55.398388
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+from chainqueue.db.migrations.sqlalchemy import (
+        chainqueue_upgrade,
+        chainqueue_downgrade,
+        )
+
+# revision identifiers, used by Alembic.
+revision = '0ec0d6d1e785'
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    chainqueue_upgrade(0, 0, 1)
+
+
+def downgrade():
+    chainqueue_downgrade(0, 0, 1)
+

apps/cic-eth/cic_eth/db/migrations/default/versions/1f1b3b641d08_roles.py

@@ -0,0 +1,29 @@
+"""Roles
+
+Revision ID: 1f1b3b641d08
+Revises: 9c420530eeb2
+Create Date: 2021-04-02 18:40:27.787631
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '1f1b3b641d08'
+down_revision = '9c420530eeb2'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'account_role',
+        sa.Column('id', sa.Integer, primary_key=True),
+        sa.Column('tag', sa.Text, nullable=False, unique=True),
+        sa.Column('address_hex', sa.String(42), nullable=False),
+        )
+
+
+def downgrade():
+    op.drop_table('account_role')

apps/cic-eth/cic_eth/db/migrations/default/versions/2a07b543335e_add_new_syncer_table.py

@@ -1,35 +0,0 @@
-"""Add new syncer table
-
-Revision ID: 2a07b543335e
-Revises: a2e2aab8f331
-Create Date: 2020-12-27 09:35:44.017981
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '2a07b543335e'
-down_revision = 'a2e2aab8f331'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'blockchain_sync',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('blockchain', sa.String, nullable=False),
-            sa.Column('block_start', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_start', sa.Integer, nullable=False, default=0),
-            sa.Column('block_cursor', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_cursor', sa.Integer, nullable=False, default=0),
-            sa.Column('block_target', sa.Integer, nullable=True),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('date_updated', sa.DateTime),
-            )
-
-
-def downgrade():
-    op.drop_table('blockchain_sync')

apps/cic-eth/cic_eth/db/migrations/default/versions/3b693afd526a_nonce_reservation.py

@@ -1,31 +0,0 @@
-"""Nonce reservation
-
-Revision ID: 3b693afd526a
-Revises: f738d9962fdf
-Create Date: 2021-03-05 07:09:50.898728
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '3b693afd526a'
-down_revision = 'f738d9962fdf'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'nonce_task_reservation',
-        sa.Column('id', sa.Integer, primary_key=True),
-        sa.Column('address_hex', sa.String(42), nullable=False),
-        sa.Column('nonce', sa.Integer, nullable=False),
-        sa.Column('key', sa.String, nullable=False),
-        sa.Column('date_created', sa.DateTime, nullable=False),
-        )
-
-
-def downgrade():
-    op.drop_table('nonce_task_reservation')

apps/cic-eth/cic_eth/db/migrations/default/versions/49b348246d70_add_nonce_index.py

@@ -1,29 +0,0 @@
-"""Add nonce index
-
-Revision ID: 49b348246d70
-Revises: 52c7c59cd0b1
-Create Date: 2020-12-19 09:45:36.186446
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '49b348246d70'
-down_revision = '52c7c59cd0b1'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-        'nonce',
-        sa.Column('id', sa.Integer, primary_key=True),
-        sa.Column('address_hex', sa.String(42), nullable=False, unique=True),
-        sa.Column('nonce', sa.Integer, nullable=False),
-        )
-
-
-def downgrade():
-    op.drop_table('nonce')

apps/cic-eth/cic_eth/db/migrations/default/versions/52c7c59cd0b1_add_account_roles.py

@@ -1,31 +0,0 @@
-"""Add account roles
-
-Revision ID: 52c7c59cd0b1
-Revises: 9c4bd7491015
-Create Date: 2020-12-19 07:21:38.249237
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '52c7c59cd0b1'
-down_revision = '9c4bd7491015'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'account_role',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('tag', sa.Text, nullable=False, unique=True),
-            sa.Column('address_hex', sa.String(42), nullable=False),
-            )
-    pass
-
-
-def downgrade():
-    op.drop_table('account_role')
-    pass

apps/cic-eth/cic_eth/db/migrations/default/versions/5ca4b77ce205_debug.py

@@ -1,8 +1,8 @@
-"""debug output
+"""DEbug
 
-Revision ID: f738d9962fdf
-Revises: ec40ac0974c1
-Create Date: 2021-03-04 08:32:43.281214
+Revision ID: 5ca4b77ce205
+Revises: 75d4767b3031
+Create Date: 2021-04-02 18:42:12.257244
 
 """
 from alembic import op
@@ -10,8 +10,8 @@ import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.
-revision = 'f738d9962fdf'
-down_revision = 'ec40ac0974c1'
+revision = '5ca4b77ce205'
+down_revision = '75d4767b3031'
 branch_labels = None
 depends_on = None
 
@@ -24,9 +24,7 @@ def upgrade():
             sa.Column('description', sa.String, nullable=False),
             sa.Column('date_created', sa.DateTime, nullable=False),
             )
-    pass
 
 
 def downgrade():
     op.drop_table('debug')
-    pass

apps/cic-eth/cic_eth/db/migrations/default/versions/6ac7a1dadc46_add_otx_state_log.py

@@ -1,30 +0,0 @@
-"""Add otx state log
-
-Revision ID: 6ac7a1dadc46
-Revises: 89e1e9baa53c
-Create Date: 2021-01-30 13:59:49.022373
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '6ac7a1dadc46'
-down_revision = '89e1e9baa53c'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'otx_state_log',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('otx_id', sa.Integer, sa.ForeignKey('otx.id'), nullable=False),
-            sa.Column('date', sa.DateTime, nullable=False),
-            sa.Column('status', sa.Integer, nullable=False),
-            )
-
-
-def downgrade():
-    op.drop_table('otx_state_log')

apps/cic-eth/cic_eth/db/migrations/default/versions/71708e943dbd_add_attempts_and_version_log_for_otx.py

@@ -1,31 +0,0 @@
-"""Add attempts and version log for otx
-
-Revision ID: 71708e943dbd
-Revises: 7e8d7626e38f
-Create Date: 2020-09-26 14:41:19.298651
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '71708e943dbd'
-down_revision = '7e8d7626e38f'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'otx_attempts',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('otx_id', sa.Integer, sa.ForeignKey('otx.id'), nullable=False),
-            sa.Column('date', sa.DateTime, nullable=False),
-            )
-    pass
-
-
-def downgrade():
-    op.drop_table('otx_attempts')
-    pass

apps/cic-eth/cic_eth/db/migrations/default/versions/75d4767b3031_lock.py

@@ -1,17 +1,20 @@
-"""Add account lock
+"""Lock
 
-Revision ID: 89e1e9baa53c
-Revises: 2a07b543335e
-Create Date: 2021-01-27 19:57:36.793882
+Revision ID: 75d4767b3031
+Revises: 1f1b3b641d08
+Create Date: 2021-04-02 18:41:20.864265
 
 """
+import datetime
 from alembic import op
 import sqlalchemy as sa
+from chainlib.eth.constant import ZERO_ADDRESS
+from cic_eth.db.enum import LockEnum
 
 
 # revision identifiers, used by Alembic.
-revision = '89e1e9baa53c'
-down_revision = '2a07b543335e'
+revision = '75d4767b3031'
+down_revision = '1f1b3b641d08'
 branch_labels = None
 depends_on = None
 
@@ -23,10 +26,12 @@ def upgrade():
             sa.Column("address", sa.String(42), nullable=True),
             sa.Column('blockchain', sa.String),
             sa.Column("flags", sa.BIGINT(), nullable=False, default=0),
-            sa.Column("date_created", sa.DateTime, nullable=False),
-            sa.Column("otx_id", sa.Integer, nullable=True),
+            sa.Column("date_created", sa.DateTime, nullable=False, default=datetime.datetime.utcnow),
+            sa.Column("otx_id", sa.Integer, sa.ForeignKey('otx.id'), nullable=True),
             )
     op.create_index('idx_chain_address', 'lock', ['blockchain', 'address'], unique=True)
+    op.execute("INSERT INTO lock (address, date_created, blockchain, flags) VALUES('{}', '{}', '::', {})".format(ZERO_ADDRESS, datetime.datetime.utcnow(), LockEnum.INIT | LockEnum.SEND | LockEnum.QUEUE))
+
 
 def downgrade():
     op.drop_index('idx_chain_address')

apps/cic-eth/cic_eth/db/migrations/default/versions/7cb65b893934_add_blocknumber_pointer.py

@@ -1,31 +0,0 @@
-"""add blocknumber pointer
-
-Revision ID: 7cb65b893934
-Revises: 8593fa1ca0f4
-Create Date: 2020-09-24 19:29:13.543648
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '7cb65b893934'
-down_revision = '8593fa1ca0f4'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'watcher_state',
-            sa.Column('block_number', sa.Integer)
-            )
-    conn = op.get_bind()
-    conn.execute('INSERT INTO watcher_state (block_number) VALUES (0);')
-    pass
-
-
-def downgrade():
-    op.drop_table('watcher_state')
-    pass

apps/cic-eth/cic_eth/db/migrations/default/versions/7e8d7626e38f_add_block_sync.py

@@ -1,45 +0,0 @@
-"""Add block sync
-
-Revision ID: 7e8d7626e38f
-Revises: cd2052be6db2
-Create Date: 2020-09-26 11:12:27.818524
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '7e8d7626e38f'
-down_revision = 'cd2052be6db2'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'block_sync',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('blockchain', sa.String, nullable=False, unique=True),
-            sa.Column('block_height_backlog', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_height_backlog', sa.Integer, nullable=False, default=0),
-            sa.Column('block_height_session', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_height_session', sa.Integer, nullable=False, default=0),
-            sa.Column('block_height_head', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_height_head', sa.Integer, nullable=False, default=0),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('date_updated', sa.DateTime),
-            )
-    op.drop_table('watcher_state')
-    pass
-
-
-def downgrade():
-    op.drop_table('block_sync')
-    op.create_table(
-            'watcher_state',
-            sa.Column('block_number', sa.Integer)
-            )
-    conn = op.get_bind()
-    conn.execute('INSERT INTO watcher_state (block_number) VALUES (0);')
-    pass

apps/cic-eth/cic_eth/db/migrations/default/versions/8593fa1ca0f4_add_transaction_queue.py

@@ -1,35 +0,0 @@
-"""Add transaction queue
-
-Revision ID: 8593fa1ca0f4
-Revises: 
-Create Date: 2020-09-22 21:56:42.117047
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '8593fa1ca0f4'
-down_revision = None
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'otx',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('nonce', sa.Integer, nullable=False),
-            sa.Column('tx_hash', sa.String(66), nullable=False),
-            sa.Column('signed_tx', sa.Text, nullable=False),
-            sa.Column('status', sa.Integer, nullable=False, default=-9),
-            sa.Column('block', sa.Integer),
-            )
-    op.create_index('idx_otx_tx', 'otx', ['tx_hash'], unique=True)
-
-
-def downgrade():
-    op.drop_index('idx_otx_tx')
-    op.drop_table('otx')

apps/cic-eth/cic_eth/db/migrations/default/versions/9c420530eeb2_nonce.py

@@ -1,8 +1,8 @@
-"""Nonce reservation
+"""Nonce
 
-Revision ID: 3b693afd526a
-Revises: f738d9962fdf
-Create Date: 2021-03-05 07:09:50.898728
+Revision ID: 9c420530eeb2
+Revises: b125cbf81e32
+Create Date: 2021-04-02 18:38:56.459334
 
 """
 from alembic import op
@@ -10,15 +10,22 @@ import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.
-revision = '3b693afd526a'
-down_revision = 'f738d9962fdf'
+revision = '9c420530eeb2'
+down_revision = 'b125cbf81e32'
 branch_labels = None
 depends_on = None
 
 
 def upgrade():
     op.create_table(
-            'nonce_task_reservation',
+        'nonce',
+        sa.Column('id', sa.Integer, primary_key=True),
+        sa.Column('address_hex', sa.String(42), nullable=False, unique=True),
+        sa.Column('nonce', sa.Integer, nullable=False),
+        )
+
+    op.create_table(
+        'nonce_task_reservation',
         sa.Column('id', sa.Integer, primary_key=True),
         sa.Column('address_hex', sa.String(42), nullable=False),
         sa.Column('nonce', sa.Integer, nullable=False),
@@ -29,3 +36,4 @@ def upgrade():
 
 def downgrade():
     op.drop_table('nonce_task_reservation')
+    op.drop_table('nonce')

apps/cic-eth/cic_eth/db/migrations/default/versions/9c4bd7491015_rename_block_sync_table.py

@@ -1,26 +0,0 @@
-"""Rename block sync table
-
-Revision ID: 9c4bd7491015
-Revises: 9daa16518a91
-Create Date: 2020-10-15 23:45:56.306898
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '9c4bd7491015'
-down_revision = '9daa16518a91'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.rename_table('block_sync', 'otx_sync')
-    pass
-
-
-def downgrade():
-    op.rename_table('otx_sync', 'block_sync')
-    pass

apps/cic-eth/cic_eth/db/migrations/default/versions/9daa16518a91_add_tx_sync_state.py

@@ -1,30 +0,0 @@
-"""add tx sync state
-
-Revision ID: 9daa16518a91
-Revises: e3b5330ee71c
-Create Date: 2020-10-10 14:43:18.699276
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '9daa16518a91'
-down_revision = 'e3b5330ee71c'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-#    op.create_table(
-#            'tx_sync',
-#            sa.Column('tx', sa.String(66), nullable=False),
-#            )
-#    op.execute("INSERT INTO tx_sync VALUES('0x0000000000000000000000000000000000000000000000000000000000000000')")
-    pass
-
-
-def downgrade():
-#    op.drop_table('tx_sync')
-    pass

apps/cic-eth/cic_eth/db/migrations/default/versions/a2e2aab8f331_add_date_accessed_to_txcache.py

@@ -1,34 +0,0 @@
-"""Add date accessed to txcache
-
-Revision ID: a2e2aab8f331
-Revises: 49b348246d70
-Create Date: 2020-12-24 18:58:06.137812
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = 'a2e2aab8f331'
-down_revision = '49b348246d70'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.add_column(
-        'tx_cache',
-        sa.Column(
-            'date_checked',
-            sa.DateTime,
-            nullable=False
-        )
-    )
-    pass
-
-
-def downgrade():
-    # drop does not work withs qlite
-    #op.drop_column('tx_cache', 'date_checked')
-    pass

apps/cic-eth/cic_eth/db/migrations/default/versions/aee12aeb47ec_convert.py

@@ -1,8 +1,8 @@
-"""convert tx index
+"""Convert
 
-Revision ID: cd2052be6db2
-Revises: 7cb65b893934
-Create Date: 2020-09-24 21:20:51.580500
+Revision ID: aee12aeb47ec
+Revises: 5ca4b77ce205
+Create Date: 2021-04-02 18:42:45.233356
 
 """
 from alembic import op
@@ -10,8 +10,8 @@ import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.
-revision = 'cd2052be6db2'
-down_revision = '7cb65b893934'
+revision = 'aee12aeb47ec'
+down_revision = '5ca4b77ce205'
 branch_labels = None
 depends_on = None
 
@@ -20,10 +20,8 @@ def upgrade():
     op.create_table(
             'tx_convert_transfer',
             sa.Column('id', sa.Integer, primary_key=True),
-            #sa.Column('approve_tx_hash', sa.String(66), nullable=False, unique=True),
             sa.Column('convert_tx_hash', sa.String(66), nullable=False, unique=True),
             sa.Column('transfer_tx_hash', sa.String(66), unique=True),
-#            sa.Column('holder_address', sa.String(42), nullable=False),
             sa.Column('recipient_address', sa.String(42), nullable=False),
             )
     op.create_index('idx_tx_convert_address', 'tx_convert_transfer', ['recipient_address'])

apps/cic-eth/cic_eth/db/migrations/default/versions/b125cbf81e32_add_chain_syncer.py

@@ -1,12 +1,13 @@
 """Add chain syncer
 
-Revision ID: ec40ac0974c1
-Revises: 6ac7a1dadc46
-Create Date: 2021-02-23 06:10:19.246304
+Revision ID: b125cbf81e32
+Revises: 0ec0d6d1e785
+Create Date: 2021-04-02 18:36:44.459603
 
 """
 from alembic import op
 import sqlalchemy as sa
+
 from chainsyncer.db.migrations.sqlalchemy import (
         chainsyncer_upgrade,
         chainsyncer_downgrade,
@@ -14,15 +15,15 @@ from chainsyncer.db.migrations.sqlalchemy import (
 
 
 # revision identifiers, used by Alembic.
-revision = 'ec40ac0974c1'
-down_revision = '6ac7a1dadc46'
+revision = 'b125cbf81e32'
+down_revision = '0ec0d6d1e785'
 branch_labels = None
 depends_on = None
 
-
 def upgrade():
     chainsyncer_upgrade(0, 0, 1)
 
 
 def downgrade():
     chainsyncer_downgrade(0, 0, 1)
+

apps/cic-eth/cic_eth/db/migrations/default/versions/df19f4e69676_add_tx_track.py

@@ -1,31 +0,0 @@
-"""Add tx tracker record
-
-Revision ID: df19f4e69676
-Revises: 71708e943dbd
-Create Date: 2020-10-09 23:31:44.563498
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-# revision identifiers, used by Alembic.
-revision = 'df19f4e69676'
-down_revision = '71708e943dbd'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-#    op.create_table(
-#            'tx',
-#            sa.Column('id', sa.Integer, primary_key=True),
-#            sa.Column('date_added', sa.DateTime, nullable=False),
-#            sa.Column('tx_hash', sa.String(66), nullable=False, unique=True),
-#            sa.Column('success', sa.Boolean(), nullable=False),
-#            )
-    pass
-
-
-def downgrade():
-#    op.drop_table('tx')
-    pass

apps/cic-eth/cic_eth/db/migrations/default/versions/e3b5330ee71c_.py

@@ -1,38 +0,0 @@
-"""Add cached values for tx
-
-Revision ID: e3b5330ee71c
-Revises: df19f4e69676
-Create Date: 2020-10-10 00:17:07.094893
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = 'e3b5330ee71c'
-down_revision = 'df19f4e69676'
-branch_labels = None
-depends_on = None
-
-def upgrade():
-    op.create_table(
-            'tx_cache',
-            sa.Column('id', sa.Integer, primary_key=True),
-#            sa.Column('tx_id', sa.Integer, sa.ForeignKey('tx.id'), nullable=True),
-            sa.Column('otx_id', sa.Integer, sa.ForeignKey('otx.id'), nullable=True),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('date_updated', sa.DateTime, nullable=False),
-            sa.Column('source_token_address', sa.String(42), nullable=False),
-            sa.Column('destination_token_address', sa.String(42), nullable=False),
-            sa.Column('sender', sa.String(42), nullable=False),
-            sa.Column('recipient', sa.String(42), nullable=False),
-            sa.Column('from_value', sa.NUMERIC(), nullable=False),
-            sa.Column('to_value', sa.NUMERIC(), nullable=True),
-            sa.Column('block_number', sa.BIGINT(), nullable=True),
-            sa.Column('tx_index', sa.Integer, nullable=True),
-            )
-
-def downgrade():
-    op.drop_table('tx_cache')
-    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/alembic.ini

@@ -1,85 +0,0 @@
-# A generic, single database configuration.
-
-[alembic]
-# path to migration scripts
-script_location = .
-
-# template used to generate migration files
-# file_template = %%(rev)s_%%(slug)s
-
-# timezone to use when rendering the date
-# within the migration file as well as the filename.
-# string value is passed to dateutil.tz.gettz()
-# leave blank for localtime
-# timezone =
-
-# max length of characters to apply to the
-# "slug" field
-# truncate_slug_length = 40
-
-# set to 'true' to run the environment during
-# the 'revision' command, regardless of autogenerate
-# revision_environment = false
-
-# set to 'true' to allow .pyc and .pyo files without
-# a source .py file to be detected as revisions in the
-# versions/ directory
-# sourceless = false
-
-# version location specification; this defaults
-# to migrations/versions.  When using multiple version
-# directories, initial revisions must be specified with --version-path
-# version_locations = %(here)s/bar %(here)s/bat migrations/versions
-
-# the output encoding used when revision files
-# are written from script.py.mako
-# output_encoding = utf-8
-
-#sqlalchemy.url = driver://user:pass@localhost/dbname
-sqlalchemy.url = postgresql+psycopg2://postgres@localhost:5432/cic-eth
-
-[post_write_hooks]
-# post_write_hooks defines scripts or Python functions that are run
-# on newly generated revision scripts.  See the documentation for further
-# detail and examples
-
-# format using "black" - use the console_scripts runner, against the "black" entrypoint
-# hooks=black
-# black.type=console_scripts
-# black.entrypoint=black
-# black.options=-l 79
-
-# Logging configuration
-[loggers]
-keys = root,sqlalchemy,alembic
-
-[handlers]
-keys = console
-
-[formatters]
-keys = generic
-
-[logger_root]
-level = WARN
-handlers = console
-qualname =
-
-[logger_sqlalchemy]
-level = WARN
-handlers =
-qualname = sqlalchemy.engine
-
-[logger_alembic]
-level = INFO
-handlers =
-qualname = alembic
-
-[handler_console]
-class = StreamHandler
-args = (sys.stderr,)
-level = NOTSET
-formatter = generic
-
-[formatter_generic]
-format = %(levelname)-5.5s [%(name)s] %(message)s
-datefmt = %H:%M:%S

apps/cic-eth/cic_eth/db/migrations/postgresql/env.py

@@ -1,77 +0,0 @@
-from logging.config import fileConfig
-
-from sqlalchemy import engine_from_config
-from sqlalchemy import pool
-
-from alembic import context
-
-# this is the Alembic Config object, which provides
-# access to the values within the .ini file in use.
-config = context.config
-
-# Interpret the config file for Python logging.
-# This line sets up loggers basically.
-fileConfig(config.config_file_name)
-
-# add your model's MetaData object here
-# for 'autogenerate' support
-# from myapp import mymodel
-# target_metadata = mymodel.Base.metadata
-target_metadata = None
-
-# other values from the config, defined by the needs of env.py,
-# can be acquired:
-# my_important_option = config.get_main_option("my_important_option")
-# ... etc.
-
-
-def run_migrations_offline():
-    """Run migrations in 'offline' mode.
-
-    This configures the context with just a URL
-    and not an Engine, though an Engine is acceptable
-    here as well.  By skipping the Engine creation
-    we don't even need a DBAPI to be available.
-
-    Calls to context.execute() here emit the given string to the
-    script output.
-
-    """
-    url = config.get_main_option("sqlalchemy.url")
-    context.configure(
-        url=url,
-        target_metadata=target_metadata,
-        literal_binds=True,
-        dialect_opts={"paramstyle": "named"},
-    )
-
-    with context.begin_transaction():
-        context.run_migrations()
-
-
-def run_migrations_online():
-    """Run migrations in 'online' mode.
-
-    In this scenario we need to create an Engine
-    and associate a connection with the context.
-
-    """
-    connectable = engine_from_config(
-        config.get_section(config.config_ini_section),
-        prefix="sqlalchemy.",
-        poolclass=pool.NullPool,
-    )
-
-    with connectable.connect() as connection:
-        context.configure(
-            connection=connection, target_metadata=target_metadata
-        )
-
-        with context.begin_transaction():
-            context.run_migrations()
-
-
-if context.is_offline_mode():
-    run_migrations_offline()
-else:
-    run_migrations_online()

apps/cic-eth/cic_eth/db/migrations/postgresql/script.py.mako

@@ -1,24 +0,0 @@
-"""${message}
-
-Revision ID: ${up_revision}
-Revises: ${down_revision | comma,n}
-Create Date: ${create_date}
-
-"""
-from alembic import op
-import sqlalchemy as sa
-${imports if imports else ""}
-
-# revision identifiers, used by Alembic.
-revision = ${repr(up_revision)}
-down_revision = ${repr(down_revision)}
-branch_labels = ${repr(branch_labels)}
-depends_on = ${repr(depends_on)}
-
-
-def upgrade():
-    ${upgrades if upgrades else "pass"}
-
-
-def downgrade():
-    ${downgrades if downgrades else "pass"}

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/2a07b543335e_add_new_syncer_table.py

@@ -1,35 +0,0 @@
-"""Add new syncer table
-
-Revision ID: 2a07b543335e
-Revises: a2e2aab8f331
-Create Date: 2020-12-27 09:35:44.017981
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '2a07b543335e'
-down_revision = 'a2e2aab8f331'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'blockchain_sync',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('blockchain', sa.String, nullable=False),
-            sa.Column('block_start', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_start', sa.Integer, nullable=False, default=0),
-            sa.Column('block_cursor', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_cursor', sa.Integer, nullable=False, default=0),
-            sa.Column('block_target', sa.Integer, nullable=True),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('date_updated', sa.DateTime),
-            )
-
-
-def downgrade():
-    op.drop_table('blockchain_sync')

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/49b348246d70_add_nonce_index.py

@@ -1,29 +0,0 @@
-"""Add nonce index
-
-Revision ID: 49b348246d70
-Revises: 52c7c59cd0b1
-Create Date: 2020-12-19 09:45:36.186446
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '49b348246d70'
-down_revision = '52c7c59cd0b1'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-        'nonce',
-        sa.Column('id', sa.Integer, primary_key=True),
-        sa.Column('address_hex', sa.String(42), nullable=False, unique=True),
-        sa.Column('nonce', sa.Integer, nullable=False),
-        )
-
-
-def downgrade():
-    op.drop_table('nonce')

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/52c7c59cd0b1_add_account_roles.py

@@ -1,31 +0,0 @@
-"""Add account roles
-
-Revision ID: 52c7c59cd0b1
-Revises: 9c4bd7491015
-Create Date: 2020-12-19 07:21:38.249237
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '52c7c59cd0b1'
-down_revision = '9c4bd7491015'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'account_role',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('tag', sa.Text, nullable=False, unique=True),
-            sa.Column('address_hex', sa.String(42), nullable=False),
-            )
-    pass
-
-
-def downgrade():
-    op.drop_table('account_role')
-    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/6ac7a1dadc46_add_otx_state_log.py

@@ -1,30 +0,0 @@
-"""Add otx state log
-
-Revision ID: 6ac7a1dadc46
-Revises: 89e1e9baa53c
-Create Date: 2021-01-30 13:59:49.022373
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '6ac7a1dadc46'
-down_revision = '89e1e9baa53c'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'otx_state_log',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('otx_id', sa.Integer, sa.ForeignKey('otx.id'), nullable=False),
-            sa.Column('date', sa.DateTime, nullable=False),
-            sa.Column('status', sa.Integer, nullable=False),
-            )
-
-
-def downgrade():
-    op.drop_table('otx_state_log')

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/71708e943dbd_add_attempts_and_version_log_for_otx.py

@@ -1,31 +0,0 @@
-"""Add attempts and version log for otx
-
-Revision ID: 71708e943dbd
-Revises: 7e8d7626e38f
-Create Date: 2020-09-26 14:41:19.298651
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '71708e943dbd'
-down_revision = '7e8d7626e38f'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'otx_attempts',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('otx_id', sa.Integer, sa.ForeignKey('otx.id'), nullable=False),
-            sa.Column('date', sa.DateTime, nullable=False),
-            )
-    pass
-
-
-def downgrade():
-    op.drop_table('otx_attempts')
-    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/7cb65b893934_add_blocknumber_pointer.py

@@ -1,31 +0,0 @@
-"""add blocknumber pointer
-
-Revision ID: 7cb65b893934
-Revises: 8593fa1ca0f4
-Create Date: 2020-09-24 19:29:13.543648
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '7cb65b893934'
-down_revision = '8593fa1ca0f4'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'watcher_state',
-            sa.Column('block_number', sa.Integer)
-            )
-    conn = op.get_bind()
-    conn.execute('INSERT INTO watcher_state (block_number) VALUES (0);')
-    pass
-
-
-def downgrade():
-    op.drop_table('watcher_state')
-    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/7e8d7626e38f_add_block_sync.py

@@ -1,42 +0,0 @@
-"""Add block sync
-
-Revision ID: 7e8d7626e38f
-Revises: cd2052be6db2
-Create Date: 2020-09-26 11:12:27.818524
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '7e8d7626e38f'
-down_revision = 'cd2052be6db2'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'block_sync',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('blockchain', sa.String, nullable=False, unique=True),
-            sa.Column('height_backlog', sa.Integer, nullable=False, default=0),
-            sa.Column('height_session', sa.Integer, nullable=False, default=0),
-            sa.Column('height_head', sa.Integer, nullable=False, default=0),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('date_updated', sa.DateTime),
-            )
-    op.drop_table('watcher_state')
-    pass
-
-
-def downgrade():
-    op.drop_table('block_sync')
-    op.create_table(
-            'watcher_state',
-            sa.Column('block_number', sa.Integer)
-            )
-    conn = op.get_bind()
-    conn.execute('INSERT INTO watcher_state (block_number) VALUES (0);')
-    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/8593fa1ca0f4_add_transaction_queue.py

@@ -1,35 +0,0 @@
-"""Add transaction queue
-
-Revision ID: 8593fa1ca0f4
-Revises: 
-Create Date: 2020-09-22 21:56:42.117047
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '8593fa1ca0f4'
-down_revision = None
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'otx',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('nonce', sa.Integer, nullable=False),
-            sa.Column('tx_hash', sa.String(66), nullable=False),
-            sa.Column('signed_tx', sa.Text, nullable=False),
-            sa.Column('status', sa.Integer, nullable=False, default=-9),
-            sa.Column('block', sa.Integer),
-            )
-    op.create_index('idx_otx_tx', 'otx', ['tx_hash'], unique=True)
-
-
-def downgrade():
-    op.drop_index('idx_otx_tx')
-    op.drop_table('otx')

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/89e1e9baa53c_add_account_lock.py

@@ -1,33 +0,0 @@
-"""Add account lock
-
-Revision ID: 89e1e9baa53c
-Revises: 2a07b543335e
-Create Date: 2021-01-27 19:57:36.793882
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '89e1e9baa53c'
-down_revision = '2a07b543335e'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'lock',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column("address", sa.String(42), nullable=True),
-            sa.Column('blockchain', sa.String),
-            sa.Column("flags", sa.BIGINT(), nullable=False, default=0),
-            sa.Column("date_created", sa.DateTime, nullable=False),
-            sa.Column("otx_id", sa.Integer, nullable=True),
-            )
-    op.create_index('idx_chain_address', 'lock', ['blockchain', 'address'], unique=True)
-
-def downgrade():
-    op.drop_index('idx_chain_address')
-    op.drop_table('lock')

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/9c4bd7491015_rename_block_sync_table.py

@@ -1,26 +0,0 @@
-"""Rename block sync table
-
-Revision ID: 9c4bd7491015
-Revises: 9daa16518a91
-Create Date: 2020-10-15 23:45:56.306898
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '9c4bd7491015'
-down_revision = '9daa16518a91'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.rename_table('block_sync', 'otx_sync')
-    pass
-
-
-def downgrade():
-    op.rename_table('otx_sync', 'block_sync')
-    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/9daa16518a91_add_tx_sync_state.py

@@ -1,30 +0,0 @@
-"""add tx sync state
-
-Revision ID: 9daa16518a91
-Revises: e3b5330ee71c
-Create Date: 2020-10-10 14:43:18.699276
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '9daa16518a91'
-down_revision = 'e3b5330ee71c'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-#    op.create_table(
-#            'tx_sync',
-#            sa.Column('tx', sa.String(66), nullable=False),
-#            )
-#    op.execute("INSERT INTO tx_sync VALUES('0x0000000000000000000000000000000000000000000000000000000000000000')")
-    pass
-
-
-def downgrade():
-#    op.drop_table('tx_sync')
-    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/a2e2aab8f331_add_date_accessed_to_txcache.py

@@ -1,33 +0,0 @@
-"""Add date accessed to txcache
-
-Revision ID: a2e2aab8f331
-Revises: 49b348246d70
-Create Date: 2020-12-24 18:58:06.137812
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = 'a2e2aab8f331'
-down_revision = '49b348246d70'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.add_column(
-        'tx_cache',
-        sa.Column(
-            'date_checked',
-            sa.DateTime,
-            nullable=False
-        )
-    )
-    pass
-
-
-def downgrade():
-    op.drop_column('tx_cache', 'date_checked')
-    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/cd2052be6db2_convert_tx_index.py

@@ -1,34 +0,0 @@
-"""convert tx index
-
-Revision ID: cd2052be6db2
-Revises: 7cb65b893934
-Create Date: 2020-09-24 21:20:51.580500
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = 'cd2052be6db2'
-down_revision = '7cb65b893934'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'tx_convert_transfer',
-            sa.Column('id', sa.Integer, primary_key=True),
-            #sa.Column('approve_tx_hash', sa.String(66), nullable=False, unique=True),
-            sa.Column('convert_tx_hash', sa.String(66), nullable=False, unique=True),
-            sa.Column('transfer_tx_hash', sa.String(66), unique=True),
-#            sa.Column('holder_address', sa.String(42), nullable=False),
-            sa.Column('recipient_address', sa.String(42), nullable=False),
-            )
-    op.create_index('idx_tx_convert_address', 'tx_convert_transfer', ['recipient_address'])
-
-
-def downgrade():
-    op.drop_index('idx_tx_convert_address')
-    op.drop_table('tx_convert_transfer')

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/df19f4e69676_add_tx_track.py

@@ -1,31 +0,0 @@
-"""Add tx tracker record
-
-Revision ID: df19f4e69676
-Revises: 71708e943dbd
-Create Date: 2020-10-09 23:31:44.563498
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-# revision identifiers, used by Alembic.
-revision = 'df19f4e69676'
-down_revision = '71708e943dbd'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-#    op.create_table(
-#            'tx',
-#            sa.Column('id', sa.Integer, primary_key=True),
-#            sa.Column('date_added', sa.DateTime, nullable=False),
-#            sa.Column('tx_hash', sa.String(66), nullable=False, unique=True),
-#            sa.Column('success', sa.Boolean(), nullable=False),
-#            )
-    pass
-
-
-def downgrade():
-#    op.drop_table('tx')
-    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/e3b5330ee71c_.py

@@ -1,37 +0,0 @@
-"""Add cached values for tx
-
-Revision ID: e3b5330ee71c
-Revises: df19f4e69676
-Create Date: 2020-10-10 00:17:07.094893
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = 'e3b5330ee71c'
-down_revision = 'df19f4e69676'
-branch_labels = None
-depends_on = None
-
-def upgrade():
-    op.create_table(
-            'tx_cache',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('otx_id', sa.Integer, sa.ForeignKey('otx.id'), nullable=True),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('date_updated', sa.DateTime, nullable=False),
-            sa.Column('source_token_address', sa.String(42), nullable=False),
-            sa.Column('destination_token_address', sa.String(42), nullable=False),
-            sa.Column('sender', sa.String(42), nullable=False),
-            sa.Column('recipient', sa.String(42), nullable=False),
-            sa.Column('from_value', sa.NUMERIC(), nullable=False),
-            sa.Column('to_value', sa.NUMERIC(), nullable=True),
-            sa.Column('block_number', sa.BIGINT(), nullable=True),
-            sa.Column('tx_index', sa.Integer, nullable=True),
-            )
-
-def downgrade():
-    op.drop_table('tx_cache')
-    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/ec40ac0974c1_add_chain_syncer.py

@@ -1,28 +0,0 @@
-"""Add chain syncer
-
-Revision ID: ec40ac0974c1
-Revises: 6ac7a1dadc46
-Create Date: 2021-02-23 06:10:19.246304
-
-"""
-from alembic import op
-import sqlalchemy as sa
-from chainsyncer.db.migrations.sqlalchemy import (
-        chainsyncer_upgrade,
-        chainsyncer_downgrade,
-        )
-
-
-# revision identifiers, used by Alembic.
-revision = 'ec40ac0974c1'
-down_revision = '6ac7a1dadc46'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    chainsyncer_upgrade(0, 0, 1)
-
-
-def downgrade():
-    chainsyncer_downgrade(0, 0, 1)

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/f738d9962fdf_debug_output.py

@@ -1,32 +0,0 @@
-"""debug output
-
-Revision ID: f738d9962fdf
-Revises: ec40ac0974c1
-Create Date: 2021-03-04 08:32:43.281214
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = 'f738d9962fdf'
-down_revision = 'ec40ac0974c1'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'debug',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('tag', sa.String, nullable=False),
-            sa.Column('description', sa.String, nullable=False),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            )
-    pass
-
-
-def downgrade():
-    op.drop_table('debug')
-    pass

apps/cic-eth/cic_eth/db/models/base.py

@@ -10,6 +10,7 @@ from sqlalchemy.pool import (
         StaticPool,
         QueuePool,
         AssertionPool,
+        NullPool,
         )
 
 logg = logging.getLogger()
@@ -64,6 +65,7 @@ class SessionBase(Model):
         if SessionBase.poolable:
             poolclass = QueuePool
             if pool_size > 1:
+                logg.info('db using queue pool')
                 e = create_engine(
                         dsn,
                         max_overflow=pool_size*3,
@@ -74,17 +76,22 @@ class SessionBase(Model):
                         echo=debug,
                     )
             else:
-                if debug:
+                if pool_size == 0:
+                    logg.info('db using nullpool')
+                    poolclass = NullPool
+                elif debug:
+                    logg.info('db using assertion pool')
                     poolclass = AssertionPool
                 else:
+                    logg.info('db using static pool')
                     poolclass = StaticPool
-
                 e = create_engine(
                         dsn,
                         poolclass=poolclass,
                         echo=debug,
                     )
         else:
+            logg.info('db not poolable')
             e = create_engine(
                     dsn,
                     echo=debug,

apps/cic-eth/cic_eth/db/models/lock.py

@@ -5,11 +5,11 @@ import logging
 # third-party imports
 from sqlalchemy import Column, String, Integer, DateTime, ForeignKey
 from chainlib.eth.constant import ZERO_ADDRESS
+from chainqueue.db.models.tx import TxCache
+from chainqueue.db.models.otx import Otx
 
 # local imports
 from cic_eth.db.models.base import SessionBase
-from cic_eth.db.models.tx import TxCache
-from cic_eth.db.models.otx import Otx
 
 logg = logging.getLogger()
 
@@ -22,10 +22,12 @@ class Lock(SessionBase):
     __tablename__ = "lock"
 
     blockchain = Column(String)
-    address = Column(String, ForeignKey('tx_cache.sender'))
+    #address = Column(String, ForeignKey('tx_cache.sender'))
+    address = Column(String, ForeignKey(TxCache.sender))
     flags = Column(Integer)
     date_created = Column(DateTime, default=datetime.datetime.utcnow)
-    otx_id = Column(Integer, ForeignKey('otx.id'))
+    otx_id = Column(Integer, ForeignKey(Otx.id))
+    #otx_id = Column(Integer)
 
 
     def chain(self):

apps/cic-eth/cic_eth/db/models/otx.py

@@ -1,680 +0,0 @@
-# standard imports
-import datetime
-import logging
-
-# external imports
-from sqlalchemy import Column, Enum, String, Integer, DateTime, Text, or_, ForeignKey
-from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
-
-# local imports
-from .base import SessionBase
-from cic_eth.db.enum import (
-        StatusEnum,
-        StatusBits,
-        status_str,
-        is_error_status,
-        )
-from cic_eth.db.error import TxStateChangeError
-
-logg = logging.getLogger()
-
-
-class OtxStateLog(SessionBase):
-
-    __tablename__ = 'otx_state_log'
-
-    date = Column(DateTime, default=datetime.datetime.utcnow)
-    status = Column(Integer)
-    otx_id = Column(Integer, ForeignKey('otx.id'))
-
-
-    def __init__(self, otx):
-        self.otx_id = otx.id
-        self.status = otx.status
-
-
-class Otx(SessionBase):
-    """Outgoing transactions with local origin.
-
-    :param nonce: Transaction nonce
-    :type nonce: number
-    :param address: Ethereum address of recipient - NOT IN USE, REMOVE
-    :type address: str
-    :param tx_hash: Tranasction hash 
-    :type tx_hash: str, 0x-hex
-    :param signed_tx: Signed raw transaction data
-    :type signed_tx: str, 0x-hex
-    """
-    __tablename__ = 'otx'
-
-    tracing = False
-    """Whether to enable queue state tracing"""
-
-    nonce = Column(Integer)
-    date_created = Column(DateTime, default=datetime.datetime.utcnow)
-    tx_hash = Column(String(66))
-    signed_tx = Column(Text)
-    status = Column(Integer)
-    block = Column(Integer)
-
-
-    def __set_status(self, status, session):
-        self.status |= status
-        session.add(self)
-        session.flush()
-
-
-    def __reset_status(self, status, session):
-        status_edit = ~status & self.status
-        self.status &= status_edit
-        session.add(self)
-        session.flush()
-   
-
-    def __status_already_set(self, status):
-        r = bool(self.status & status)
-        if r:
-            logg.warning('status bit {} already set on {}'.format(status.name, self.tx_hash))
-        return r
-
-
-    def __status_not_set(self, status):
-        r = not(self.status & status)
-        if r:
-            logg.warning('status bit {} not set on {}'.format(status.name, self.tx_hash))
-        return r
-
-
-    def set_block(self, block, session=None):
-        """Set block number transaction was mined in.
-
-        Only manipulates object, does not transaction or commit to backend.
-
-        :param block: Block number
-        :type block: number
-        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
-        """
-        session = SessionBase.bind_session(session)
-
-        if self.block != None:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('Attempted set block {} when block was already {}'.format(block, self.block))
-        self.block = block
-        session.add(self)
-        session.flush()
-
-        SessionBase.release_session(session)
-
-
-    def waitforgas(self, session=None):
-        """Marks transaction as suspended pending gas funding.
-
-        Only manipulates object, does not transaction or commit to backend.
-
-        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
-        """
-        if self.__status_already_set(StatusBits.GAS_ISSUES):
-            return
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('GAS_ISSUES cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-        if self.status & StatusBits.IN_NETWORK:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('GAS_ISSUES cannot be set on an entry with IN_NETWORK state set ({})'.format(status_str(self.status)))
-
-        self.__set_status(StatusBits.GAS_ISSUES, session)
-        self.__reset_status(StatusBits.QUEUED | StatusBits.DEFERRED, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    def fubar(self, session=None):
-        """Marks transaction as "fubar." Any transaction marked this way is an anomaly and may be a symptom of a serious problem.
-
-        Only manipulates object, does not transaction or commit to backend.
-        """
-        if self.__status_already_set(StatusBits.UNKNOWN_ERROR):
-            return
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('FUBAR cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-        if is_error_status(self.status):
-            SessionBase.release_session(session)
-            raise TxStateChangeError('FUBAR cannot be set on an entry with an error state already set ({})'.format(status_str(self.status)))
-
-        self.__set_status(StatusBits.UNKNOWN_ERROR | StatusBits.FINAL, session)
-       
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    def reject(self, session=None):
-        """Marks transaction as "rejected," which means the node rejected sending the transaction to the network. The nonce has not been spent, and the transaction should be replaced.
-
-        Only manipulates object, does not transaction or commit to backend.
-        """
-        if self.__status_already_set(StatusBits.NODE_ERROR):
-            return
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('REJECTED cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-        if self.status & StatusBits.IN_NETWORK:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('REJECTED cannot be set on an entry already IN_NETWORK ({})'.format(status_str(self.status)))
-        if is_error_status(self.status):
-            SessionBase.release_session(session)
-            raise TxStateChangeError('REJECTED cannot be set on an entry with an error state already set ({})'.format(status_str(self.status)))
-
-        self.__set_status(StatusBits.NODE_ERROR | StatusBits.FINAL, session)
-            
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    def override(self, manual=False, session=None):
-        """Marks transaction as manually overridden.
-
-        Only manipulates object, does not transaction or commit to backend.
-        """
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('OVERRIDDEN/OBSOLETED cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-        if self.status & StatusBits.IN_NETWORK:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('OVERRIDDEN/OBSOLETED cannot be set on an entry already IN_NETWORK ({})'.format(status_str(self.status)))
-        if self.status & StatusBits.OBSOLETE:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('OVERRIDDEN/OBSOLETED cannot be set on an entry already OBSOLETE ({})'.format(status_str(self.status)))
-
-        self.__set_status(StatusBits.OBSOLETE, session)
-        #if manual:
-        #    self.__set_status(StatusBits.MANUAL, session)
-        self.__reset_status(StatusBits.QUEUED | StatusBits.IN_NETWORK, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    def manual(self, session=None):
-
-        session = SessionBase.bind_session(session)
-
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('OVERRIDDEN/OBSOLETED cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-
-        self.__set_status(StatusBits.MANUAL, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-    def retry(self, session=None):
-        """Marks transaction as ready to retry after a timeout following a sendfail or a completed gas funding.
-
-        Only manipulates object, does not transaction or commit to backend.
-
-        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
-        """
-        if self.__status_already_set(StatusBits.QUEUED):
-            return
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('RETRY cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-        if not is_error_status(self.status) and not StatusBits.IN_NETWORK & self.status > 0:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('RETRY cannot be set on an entry that has no error ({})'.format(status_str(self.status)))
-
-        self.__set_status(StatusBits.QUEUED, session)
-        self.__reset_status(StatusBits.GAS_ISSUES, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    def readysend(self, session=None):
-        """Marks transaction as ready for initial send attempt.
-
-        Only manipulates object, does not transaction or commit to backend.
-
-        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
-        """
-        if self.__status_already_set(StatusBits.QUEUED):
-            return
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('READYSEND cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-        if is_error_status(self.status):
-            SessionBase.release_session(session)
-            raise TxStateChangeError('READYSEND cannot be set on an errored state ({})'.format(status_str(self.status)))
-
-        self.__set_status(StatusBits.QUEUED, session)
-        self.__reset_status(StatusBits.GAS_ISSUES, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    def sent(self, session=None):
-        """Marks transaction as having been sent to network.
-
-        Only manipulates object, does not transaction or commit to backend.
-
-        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
-        """
-        if self.__status_already_set(StatusBits.IN_NETWORK):
-            return
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('SENT cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-
-        self.__set_status(StatusBits.IN_NETWORK, session)
-        self.__reset_status(StatusBits.DEFERRED | StatusBits.QUEUED | StatusBits.LOCAL_ERROR | StatusBits.NODE_ERROR, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    def sendfail(self, session=None):
-        """Marks that an attempt to send the transaction to the network has failed.
-
-        Only manipulates object, does not transaction or commit to backend.
-
-        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
-        """
-        if self.__status_already_set(StatusBits.NODE_ERROR):
-            return
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('SENDFAIL cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-        if self.status & StatusBits.IN_NETWORK:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('SENDFAIL cannot be set on an entry with IN_NETWORK state set ({})'.format(status_str(self.status)))
-
-        self.__set_status(StatusBits.LOCAL_ERROR | StatusBits.DEFERRED, session)
-        self.__reset_status(StatusBits.QUEUED | StatusBits.GAS_ISSUES, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    def dequeue(self, session=None):
-        """Marks that a process to execute send attempt is underway
-
-        Only manipulates object, does not transaction or commit to backend.
-
-        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
-        """
-        if self.__status_not_set(StatusBits.QUEUED):
-            return
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('QUEUED cannot be unset on an entry with FINAL state set ({})'.format(status_str(self.status)))
-        if self.status & StatusBits.IN_NETWORK:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('QUEUED cannot be unset on an entry with IN_NETWORK state set ({})'.format(status_str(self.status)))
-
-        self.__reset_status(StatusBits.QUEUED, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-
-    def minefail(self, block, session=None):
-        """Marks that transaction was mined but code execution did not succeed.
-
-        Only manipulates object, does not transaction or commit to backend.
-
-        :param block: Block number transaction was mined in.
-        :type block: number
-        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
-        """
-        if self.__status_already_set(StatusBits.NETWORK_ERROR):
-            return
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('REVERTED cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-        if not self.status & StatusBits.IN_NETWORK:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('REVERTED cannot be set on an entry without IN_NETWORK state set ({})'.format(status_str(self.status)))
-
-        if block != None:
-            self.block = block
-
-        self.__set_status(StatusBits.NETWORK_ERROR | StatusBits.FINAL, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    def cancel(self, confirmed=False, session=None):
-        """Marks that the transaction has been succeeded by a new transaction with same nonce.
-
-        If set to confirmed, the previous state must be OBSOLETED, and will transition to CANCELLED - a finalized state. Otherwise, the state must follow a non-finalized state, and will be set to OBSOLETED.
-
-        Only manipulates object, does not transaction or commit to backend.
-
-        :param confirmed: Whether transition is to a final state.
-        :type confirmed: bool
-        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
-        """
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('CANCEL cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-
-        if confirmed:
-            if self.status > 0 and not self.status & StatusBits.OBSOLETE:
-                SessionBase.release_session(session)
-                raise TxStateChangeError('CANCEL can only be set on an entry marked OBSOLETE ({})'.format(status_str(self.status)))
-            self.__set_status(StatusEnum.CANCELLED, session)
-        else:
-            self.__set_status(StatusEnum.OBSOLETED, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    def success(self, block, session=None):
-        """Marks that transaction was successfully mined.
-
-        Only manipulates object, does not transaction or commit to backend.
-
-        :param block: Block number transaction was mined in.
-        :type block: number
-        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
-        """
-
-        session = SessionBase.bind_session(session)
-
-        if self.status & StatusBits.FINAL:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('SUCCESS cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
-        if not self.status & StatusBits.IN_NETWORK:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('SUCCESS cannot be set on an entry without IN_NETWORK state set ({})'.format(status_str(self.status)))
-        if is_error_status(self.status):
-            SessionBase.release_session(session)
-            raise TxStateChangeError('SUCCESS cannot be set on an entry with error state set ({})'.format(status_str(self.status)))
-
-        if block != None:
-            self.block = block
-        self.__set_status(StatusEnum.SUCCESS, session)
-
-        if self.tracing:
-            self.__state_log(session=session)
-
-        SessionBase.release_session(session)
-
-
-    @staticmethod
-    def get(status=0, limit=4096, status_exact=True, session=None):
-        """Returns outgoing transaction lists by status.
-
-        Status may either be matched exactly, or be an upper bound of the integer value of the status enum.
-
-        :param status: Status value to use in query
-        :type status: cic_eth.db.enum.StatusEnum
-        :param limit: Max results to return
-        :type limit: number
-        :param status_exact: Whether or not to perform exact status match
-        :type bool:
-        :returns: List of transaction hashes
-        :rtype: tuple, where first element is transaction hash
-        """
-        e = None
-
-        session = SessionBase.bind_session(session)
-
-        if status_exact:
-            e = session.query(Otx.tx_hash).filter(Otx.status==status).order_by(Otx.date_created.asc()).limit(limit).all()
-        else:
-            e = session.query(Otx.tx_hash).filter(Otx.status<=status).order_by(Otx.date_created.asc()).limit(limit).all()
-        
-        SessionBase.release_session(session)
-        return e
-
-
-    @staticmethod
-    def load(tx_hash, session=None):
-        """Retrieves the outgoing transaction record by transaction hash.
-
-        :param tx_hash: Transaction hash
-        :type tx_hash: str, 0x-hex
-        """
-        session = SessionBase.bind_session(session)
-        
-        q = session.query(Otx)
-        q = q.filter(Otx.tx_hash==tx_hash)
-
-        SessionBase.release_session(session)
-
-        return q.first()
-
-
-    @staticmethod
-    def account(account_address):
-        """Retrieves all transaction hashes for which the given Ethereum address is sender or recipient.
-
-        :param account_address: Ethereum address to use in query.
-        :type account_address: str, 0x-hex
-        :returns: Outgoing transactions
-        :rtype: tuple, where first element is transaction hash
-        """
-        session = Otx.create_session()
-        q = session.query(Otx.tx_hash)
-        q = q.join(TxCache)
-        q = q.filter(or_(TxCache.sender==account_address, TxCache.recipient==account_address))
-        txs = q.all()
-        session.close()
-        return list(txs)
-
-
-    def __state_log(self, session):
-        l = OtxStateLog(self)
-        session.add(l)
-
-
-    # TODO: it is not safe to return otx here unless session has been passed in
-    @staticmethod
-    def add(nonce, address, tx_hash, signed_tx, session=None):
-        external_session = session != None
-
-        session = SessionBase.bind_session(session)
-
-        otx = Otx(nonce, address, tx_hash, signed_tx)
-        session.add(otx)
-        session.flush()
-        if otx.tracing:
-            otx.__state_log(session=session)
-        session.flush()
-
-        SessionBase.release_session(session)
-        
-        if not external_session:
-            return None
-
-        return otx
-
-
-    def __init__(self, nonce, address, tx_hash, signed_tx):
-        self.nonce = nonce
-        self.tx_hash = tx_hash
-        self.signed_tx = signed_tx
-        self.status = StatusEnum.PENDING
-        signed_tx_bytes = bytes.fromhex(signed_tx[2:])
-
-       # sender_address = address_hex_from_signed_tx(signed_tx_bytes)
-       # logg.debug('decoded tx {}'.format(sender_address))
-
-    
-
-# TODO: Most of the methods on this object are obsolete, but it contains a static function for retrieving "expired" outgoing transactions that should be moved to Otx instead.
-class OtxSync(SessionBase):
-    """Obsolete
-    """
-    __tablename__ = 'otx_sync'
-
-    blockchain = Column(String)
-    block_height_backlog = Column(Integer)
-    tx_height_backlog = Column(Integer)
-    block_height_session = Column(Integer)
-    tx_height_session = Column(Integer)
-    block_height_head = Column(Integer)
-    tx_height_head = Column(Integer)
-    date_created = Column(DateTime, default=datetime.datetime.utcnow)
-    date_updated = Column(DateTime)
-
-
-    def backlog(self, block_height=None, tx_height=None):
-        #session = OtxSync.create_session()
-        if block_height != None:
-            if tx_height == None:
-                raise ValueError('tx height missing')
-            self.block_height_backlog = block_height
-            self.tx_height_backlog = tx_height
-            #session.add(self)
-            self.date_updated = datetime.datetime.utcnow()
-        #session.commit()
-        block_height = self.block_height_backlog
-        tx_height = self.tx_height_backlog
-        #session.close()
-        return (block_height, tx_height)
-   
-
-    def session(self, block_height=None, tx_height=None):
-        #session = OtxSync.create_session()
-        if block_height != None:
-            if tx_height == None:
-                raise ValueError('tx height missing')
-            self.block_height_session = block_height
-            self.tx_height_session = tx_height
-            #session.add(self)
-            self.date_updated = datetime.datetime.utcnow()
-        #session.commit()
-        block_height = self.block_height_session
-        tx_height = self.tx_height_session
-        #session.close()
-        return (block_height, tx_height)
-
-
-    def head(self, block_height=None, tx_height=None):
-        #session = OtxSync.create_session()
-        if block_height != None:
-            if tx_height == None:
-                raise ValueError('tx height missing')
-            self.block_height_head = block_height
-            self.tx_height_head = tx_height
-            #session.add(self)
-            self.date_updated = datetime.datetime.utcnow()
-        #session.commit()
-        block_height = self.block_height_head
-        tx_height = self.tx_height_head
-        #session.close()
-        return (block_height, tx_height)
-
-
-    @hybrid_property
-    def synced(self):
-        #return self.block_height_session == self.block_height_backlog and self.tx_height_session == self.block_height_backlog
-        return self.block_height_session == self.block_height_backlog and self.tx_height_session == self.tx_height_backlog
-
-
-    @staticmethod
-    def load(blockchain_string, session):
-        q = session.query(OtxSync)
-        q = q.filter(OtxSync.blockchain==blockchain_string)
-        return q.first()
-
-
-    @staticmethod
-    def latest(nonce):
-        session = SessionBase.create_session()
-        otx = session.query(Otx).filter(Otx.nonce==nonce).order_by(Otx.created.desc()).first()
-        session.close()
-        return otx
-
-
-    @staticmethod
-    def get_expired(datetime_threshold):
-        session = SessionBase.create_session()
-        q = session.query(Otx)
-        q = q.filter(Otx.date_created<datetime_threshold)
-        q = q.filter(Otx.status==StatusEnum.SENT)
-        q = q.order_by(Otx.date_created.desc())
-        q = q.group_by(Otx.nonce)
-        q = q.group_by(Otx.id)
-        otxs = q.all()
-        session.close()
-        return otxs
-
-
-    def chain(self):
-        return self.blockchain
-
-
-    def __init__(self, blockchain):
-        self.blockchain = blockchain
-        self.block_height_head = 0
-        self.tx_height_head = 0
-        self.block_height_session = 0
-        self.tx_height_session = 0
-        self.block_height_backlog = 0
-        self.tx_height_backlog = 0

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

@@ -1,151 +0,0 @@
-# standard imports
-import datetime
-
-# third-party imports
-from sqlalchemy import Column, String, Integer, DateTime, Enum, ForeignKey, Boolean, NUMERIC
-from sqlalchemy.ext.hybrid import hybrid_method, hybrid_property
-#from sqlalchemy.orm import relationship, backref
-#from sqlalchemy.ext.declarative import declarative_base
-
-# local imports
-from .base import SessionBase
-from .otx import Otx
-from cic_eth.db.util import num_serialize
-from cic_eth.error import NotLocalTxError
-from cic_eth.db.error import TxStateChangeError
-
-
-class TxCache(SessionBase):
-    """Metadata expansions for outgoing transactions.
-
-    These records are not essential for handling of outgoing transaction queues. It is implemented to reduce the amount of computation spent of parsing and analysing raw signed transaction data.
-
-    Instantiation of the object will fail if an outgoing transaction record with the same transaction hash does not exist.
-
-    Typically three types of transactions are recorded:
-
-    - Token transfers; where source and destination token values and addresses are identical, sender and recipient differ.
-    - Token conversions; source and destination token values and addresses differ, sender and recipient are identical.
-    - Any other transaction; source and destination token addresses are zero-address.
-
-    :param tx_hash: Transaction hash
-    :type tx_hash: str, 0x-hex
-    :param sender: Ethereum address of transaction sender
-    :type sender: str, 0x-hex
-    :param recipient: Ethereum address of transaction beneficiary (e.g. token transfer recipient)
-    :type recipient: str, 0x-hex
-    :param source_token_address: Contract address of token that sender spent from
-    :type source_token_address: str, 0x-hex
-    :param destination_token_address: Contract address of token that recipient will receive balance of
-    :type destination_token_address: str, 0x-hex
-    :param from_value: Amount of source tokens spent
-    :type from_value: number
-    :param to_value: Amount of destination tokens received
-    :type to_value: number
-    :param block_number: Block height the transaction was mined at, or None if not yet mined
-    :type block_number: number or None
-    :param tx_number: Block transaction height the transaction was mined at, or None if not yet mined
-    :type tx_number: number or None
-    :raises FileNotFoundError: Outgoing transaction for given transaction hash does not exist
-    """
-    __tablename__ = 'tx_cache'
-
-    otx_id = Column(Integer, ForeignKey('otx.id'))
-    source_token_address = Column(String(42))
-    destination_token_address = Column(String(42))
-    sender = Column(String(42))
-    recipient = Column(String(42))
-    from_value = Column(NUMERIC())
-    to_value = Column(NUMERIC())
-    block_number = Column(Integer())
-    tx_index = Column(Integer())
-    date_created = Column(DateTime, default=datetime.datetime.utcnow)
-    date_updated = Column(DateTime, default=datetime.datetime.utcnow)
-    date_checked = Column(DateTime, default=datetime.datetime.utcnow)
-
-
-    def check(self):
-        """Update the "checked" timestamp to current time.
-
-        Only manipulates object, does not transaction or commit to backend.
-        """
-        self.date_checked = datetime.datetime.now()
-
-
-    @staticmethod
-    def clone(
-            tx_hash_original,
-            tx_hash_new,
-            session=None,
-            ):
-        """Copy tx cache data and associate it with a new transaction.
-
-        :param tx_hash_original: tx cache data to copy
-        :type tx_hash_original: str, 0x-hex
-        :param tx_hash_new: tx hash to associate the copied entry with
-        :type tx_hash_new: str, 0x-hex
-        """
-        session = SessionBase.bind_session(session)
-        
-        q = session.query(TxCache)
-        q = q.join(Otx)
-        q = q.filter(Otx.tx_hash==tx_hash_original)
-        txc = q.first()
-
-        if txc == None:
-            SessionBase.release_session(session)
-            raise NotLocalTxError('original {}'.format(tx_hash_original))
-        if txc.block_number != None:
-            SessionBase.release_session(session)
-            raise TxStateChangeError('cannot clone tx cache of confirmed tx {}'.format(tx_hash_original))
-
-        session.flush()
-        q = session.query(Otx)
-        q = q.filter(Otx.tx_hash==tx_hash_new)
-        otx = q.first()
-
-        if otx == None:
-            SessionBase.release_session(session)
-            raise NotLocalTxError('new {}'.format(tx_hash_new))
-
-        txc_new = TxCache(
-                otx.tx_hash,
-                txc.sender,
-                txc.recipient,
-                txc.source_token_address,
-                txc.destination_token_address,
-                int(txc.from_value),
-                int(txc.to_value),
-                session=session,
-                )
-        session.add(txc_new)
-        session.commit()
-
-        SessionBase.release_session(session)
-
-
-    def __init__(self, tx_hash, sender, recipient, source_token_address, destination_token_address, from_value, to_value, block_number=None, tx_index=None, session=None):
-        session = SessionBase.bind_session(session)
-        q = session.query(Otx)
-        q = q.filter(Otx.tx_hash==tx_hash)
-        tx = q.first()
-        if tx == None:
-            SessionBase.release_session(session)
-            raise FileNotFoundError('outgoing transaction record unknown {} (add a Tx first)'.format(tx_hash))
-        self.otx_id = tx.id
-
-        self.sender = sender
-        self.recipient = recipient
-        self.source_token_address = source_token_address
-        self.destination_token_address = destination_token_address
-        self.from_value = from_value
-        self.to_value = to_value
-        self.block_number = block_number
-        self.tx_index = tx_index
-        # not automatically set in sqlite, it seems:
-        self.date_created = datetime.datetime.utcnow()
-        self.date_updated = self.date_created
-        self.date_checked = self.date_created
-
-        SessionBase.release_session(session)
-

apps/cic-eth/cic_eth/db/util.py

@@ -1,8 +0,0 @@
-import math
-
-def num_serialize(n):
-    if n == 0:
-        return b'\x00'
-    binlog = math.log2(n)
-    bytelength = int(binlog / 8 + 1)
-    return n.to_bytes(bytelength, 'big')

apps/cic-eth/cic_eth/error.py

@@ -4,12 +4,6 @@ class TokenCountError(Exception):
     pass
 
 
-class NotLocalTxError(Exception):
-    """Exception raised when trying to access a tx not originated from a local task
-    """
-    pass
-
-
 class PermanentTxError(Exception):
     """Exception raised when encountering a permanent error when sending a tx.
 
@@ -54,6 +48,8 @@ class RoleMissingError(Exception):
     pass
 
 
+
+
 class IntegrityError(Exception):
     """Exception raised to signal irregularities with deduplication and ordering of tasks
 
@@ -68,15 +64,19 @@ class LockedError(Exception):
     pass
 
 
-class SignerError(Exception):
+class SeppukuError(Exception):
+    """Exception base class for all errors that should cause system shutdown
+
+    """
+
+
+class SignerError(SeppukuError):
     """Exception raised when signer is unavailable or generates an error
 
     """
     pass
 
 
-class EthError(Exception):
-    """Exception raised when unspecified error from evm node is encountered
-
+class RoleAgencyError(SeppukuError):
+    """Exception raise when a role cannot perform its function. This is a critical exception
     """
-    pass

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

@@ -3,11 +3,11 @@ import logging
 
 # external imports
 import celery
-from erc20_single_shot_faucet import SingleShotFaucet as Faucet
-from chainlib.eth.constant import ZERO_ADDRESS
+from erc20_faucet import Faucet
 from hexathon import (
         strip_0x,
         )
+from chainlib.eth.constant import ZERO_ADDRESS
 from chainlib.connection import RPCConnection
 from chainlib.eth.sign import (
         new_account,
@@ -19,8 +19,11 @@ from chainlib.eth.tx import (
         unpack,
         )
 from chainlib.chain import ChainSpec
-from eth_accounts_index import AccountRegistry
-from sarafu_faucet import MinterFaucet as Faucet
+from chainlib.error import JSONRPCException
+from eth_accounts_index.registry import AccountRegistry
+from eth_accounts_index import AccountsIndex 
+from sarafu_faucet import MinterFaucet
+from chainqueue.db.models.tx import TxCache
 
 # local import
 from cic_eth_registry import CICRegistry
@@ -31,7 +34,6 @@ from cic_eth.eth.gas import (
 from cic_eth.db.models.nonce import Nonce
 from cic_eth.db.models.base import SessionBase
 from cic_eth.db.models.role import AccountRole
-from cic_eth.db.models.tx import TxCache
 from cic_eth.error import (
         RoleMissingError,
         SignerError,
@@ -70,11 +72,18 @@ def create(self, password, chain_spec_dict):
     a = None
     conn = RPCConnection.connect(chain_spec, 'signer')
     o = new_account()
-    a = conn.do(o)
+    try:
+        a = conn.do(o)
+    except ConnectionError as e:
+        raise SignerError(e)
+    except FileNotFoundError as e:
+        raise SignerError(e)
     conn.disconnect()
 
+    # TODO: It seems infeasible that a can be None in any case, verify
     if a == None:
         raise SignerError('create account')
+        
     logg.debug('created account {}'.format(a))
 
     # Initialize nonce provider record for account
@@ -125,7 +134,7 @@ def register(self, account_address, chain_spec_dict, writer_address=None):
     rpc_signer = RPCConnection.connect(chain_spec, 'signer')
     nonce_oracle = CustodialTaskNonceOracle(writer_address, self.request.root_id, session=session) #, default_nonce)
     gas_oracle = self.create_gas_oracle(rpc, AccountRegistry.gas)
-    account_registry = AccountRegistry(signer=rpc_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle, chain_id=chain_spec.chain_id())
+    account_registry = AccountsIndex(chain_spec, signer=rpc_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
     (tx_hash_hex, tx_signed_raw_hex) = account_registry.add(account_registry_address, writer_address, account_address, tx_format=TxFormat.RLP_SIGNED)
     rpc_signer.disconnect()
 
@@ -177,8 +186,8 @@ def gift(self, account_address, chain_spec_dict):
     # Generate and sign transaction
     rpc_signer = RPCConnection.connect(chain_spec, 'signer')
     nonce_oracle = CustodialTaskNonceOracle(account_address, self.request.root_id, session=session) #, default_nonce)
-    gas_oracle = self.create_gas_oracle(rpc, Faucet.gas)
-    faucet = Faucet(signer=rpc_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle, chain_id=chain_spec.chain_id())
+    gas_oracle = self.create_gas_oracle(rpc, MinterFaucet.gas)
+    faucet = Faucet(chain_spec, signer=rpc_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
     (tx_hash_hex, tx_signed_raw_hex) = faucet.give_to(faucet_address, account_address, account_address, tx_format=TxFormat.RLP_SIGNED)
     rpc_signer.disconnect()
 
@@ -219,21 +228,22 @@ def have(self, account, chain_spec_dict):
     """
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     o = sign_message(account, '0x2a')
-    try:
-        conn = RPCConnection.connect(chain_spec, 'signer')
-    except Exception as e:
-        logg.debug('cannot sign with {}: {}'.format(account, e))
-        return None
+    conn = RPCConnection.connect(chain_spec, 'signer')
 
     try:
         conn.do(o)
-        conn.disconnect()
-        return account
-    except Exception as e:
+    except ConnectionError as e:
+        raise SignerError(e)
+    except FileNotFoundError as e:
+        raise SignerError(e)
+    except JSONRPCException as e:
         logg.debug('cannot sign with {}: {}'.format(account, e))
         conn.disconnect()
         return None
    
+    conn.disconnect()
+    return account
+
 
 @celery_app.task(bind=True, base=CriticalSQLAlchemyTask)
 def set_role(self, tag, address, chain_spec_dict):
@@ -285,7 +295,7 @@ def cache_gift_data(
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
 
     tx_signed_raw_bytes = bytes.fromhex(strip_0x(tx_signed_raw_hex))
-    tx = unpack(tx_signed_raw_bytes, chain_spec.chain_id())
+    tx = unpack(tx_signed_raw_bytes, chain_spec)
     tx_data = Faucet.parse_give_to_request(tx['data'])
 
     session = self.create_session()
@@ -328,8 +338,8 @@ def cache_account_data(
     """
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     tx_signed_raw_bytes = bytes.fromhex(tx_signed_raw_hex[2:])
-    tx = unpack(tx_signed_raw_bytes, chain_id=chain_spec.chain_id())
-    tx_data = AccountRegistry.parse_add_request(tx['data'])
+    tx = unpack(tx_signed_raw_bytes, chain_spec)
+    tx_data = AccountsIndex.parse_add_request(tx['data'])
 
     session = SessionBase.create_session()
     tx_cache = TxCache(

apps/cic-eth/cic_eth/eth/bancor.py.bak

@@ -209,7 +209,7 @@ def convert_with_default_reserve(self, tokens, from_address, source_amount, mini
 #    s_queue.apply_async()
 #
 #    s_check_gas = celery.signature(
-#            'cic_eth.eth.tx.check_gas',
+#            'cic_eth.eth.gas.check_gas',
 #            [
 #                c['address'],
 #                [c['signed_tx']],
@@ -222,7 +222,7 @@ def convert_with_default_reserve(self, tokens, from_address, source_amount, mini
 #            )
 #
 #    s_set_sent = celery.signature(
-#            'cic_eth.queue.tx.set_sent_status',
+#            'cic_eth.queue.state.set_sent',
 #            [False],
 #            )
 #    s_send.link(s_set_sent)
@@ -364,7 +364,7 @@ def otx_cache_convert(
 
     chain_spec = ChainSpec.from_chain_str(chain_str)
     tx_signed_raw_bytes = bytes.fromhex(tx_signed_raw_hex[2:])
-    tx = unpack_signed_raw_tx(tx_signed_raw_bytes, chain_spec.chain_id())
+    tx = unpack(tx_signed_raw_bytes, chain_spec)
     tx_data = unpack_convert(tx['data'])
     logg.debug('tx data {}'.format(tx_data))
 

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

@@ -6,7 +6,6 @@ import celery
 from chainlib.eth.constant import ZERO_ADDRESS
 from chainlib.chain import ChainSpec
 from chainlib.connection import RPCConnection
-from chainlib.eth.erc20 import ERC20
 from chainlib.eth.tx import (
         TxFormat,
         unpack,
@@ -14,18 +13,23 @@ from chainlib.eth.tx import (
 from cic_eth_registry import CICRegistry
 from cic_eth_registry.erc20 import ERC20Token
 from hexathon import strip_0x
+from chainqueue.db.models.tx import TxCache
+from chainqueue.error import NotLocalTxError
+from eth_erc20 import ERC20
 
 # local imports
-from cic_eth.db.models.tx import TxCache
 from cic_eth.db.models.base import SessionBase
 from cic_eth.db.models.role import AccountRole
-from cic_eth.error import TokenCountError, PermanentTxError, OutOfGasError, NotLocalTxError
+from cic_eth.error import (
+        TokenCountError,
+        PermanentTxError,
+        OutOfGasError,
+        )
 from cic_eth.queue.tx import register_tx
 from cic_eth.eth.gas import (
         create_check_gas_task,
         MaxGasOracle,
         )
-#from cic_eth.eth.factory import TxFactory
 from cic_eth.ext.address import translate_address
 from cic_eth.task import (
         CriticalSQLAlchemyTask,
@@ -57,8 +61,8 @@ def balance(tokens, holder_address, chain_spec_dict):
 
     for t in tokens:
         address = t['address']
-        token = ERC20Token(rpc, address)
-        c = ERC20()
+        token = ERC20Token(chain_spec, rpc, address)
+        c = ERC20(chain_spec)
         o = c.balance_of(address, holder_address, sender_address=caller_address)
         r = rpc.do(o)
         t['balance_network'] = c.parse_balance(r)
@@ -90,6 +94,7 @@ def transfer(self, tokens, holder_address, receiver_address, value, chain_spec_d
     :rtype: str, 0x-hex
     """
     # we only allow one token, one transfer
+    logg.debug('tokens {}'.format(tokens))
     if len(tokens) != 1:
         raise TokenCountError
     t = tokens[0]
@@ -102,8 +107,14 @@ def transfer(self, tokens, holder_address, receiver_address, value, chain_spec_d
     session = self.create_session()
     nonce_oracle = CustodialTaskNonceOracle(holder_address, self.request.root_id, session=session)
     gas_oracle = self.create_gas_oracle(rpc, MaxGasOracle.gas)
-    c = ERC20(signer=rpc_signer, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle, chain_id=chain_spec.chain_id())
-    (tx_hash_hex, tx_signed_raw_hex) = c.transfer(t['address'], holder_address, receiver_address, value, tx_format=TxFormat.RLP_SIGNED)
+    c = ERC20(chain_spec, signer=rpc_signer, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle)
+    try:
+        (tx_hash_hex, tx_signed_raw_hex) = c.transfer(t['address'], holder_address, receiver_address, value, tx_format=TxFormat.RLP_SIGNED)
+    except FileNotFoundError as e:
+        raise SignerError(e)
+    except ConnectionError as e:
+        raise SignerError(e)
+    
 
     rpc_signer.disconnect()
     rpc.disconnect()
@@ -165,8 +176,13 @@ def approve(self, tokens, holder_address, spender_address, value, chain_spec_dic
     session = self.create_session()
     nonce_oracle = CustodialTaskNonceOracle(holder_address, self.request.root_id, session=session)
     gas_oracle = self.create_gas_oracle(rpc, MaxGasOracle.gas)
-    c = ERC20(signer=rpc_signer, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle, chain_id=chain_spec.chain_id())
-    (tx_hash_hex, tx_signed_raw_hex) = c.approve(t['address'], holder_address, spender_address, value, tx_format=TxFormat.RLP_SIGNED)
+    c = ERC20(chain_spec, signer=rpc_signer, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle)
+    try:
+        (tx_hash_hex, tx_signed_raw_hex) = c.approve(t['address'], holder_address, spender_address, value, tx_format=TxFormat.RLP_SIGNED)
+    except FileNotFoundError as e:
+        raise SignerError(e)
+    except ConnectionError as e:
+        raise SignerError(e)
 
     rpc_signer.disconnect()
     rpc.disconnect()
@@ -239,7 +255,7 @@ def cache_transfer_data(
     """
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     tx_signed_raw_bytes = bytes.fromhex(strip_0x(tx_signed_raw_hex))
-    tx = unpack(tx_signed_raw_bytes, chain_spec.chain_id())
+    tx = unpack(tx_signed_raw_bytes, chain_spec)
 
     tx_data = ERC20.parse_transfer_request(tx['data'])
     recipient_address = tx_data[0]
@@ -280,7 +296,7 @@ def cache_approve_data(
     """
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     tx_signed_raw_bytes = bytes.fromhex(strip_0x(tx_signed_raw_hex))
-    tx = unpack(tx_signed_raw_bytes, chain_spec.chain_id())
+    tx = unpack(tx_signed_raw_bytes, chain_spec)
 
     tx_data = ERC20.parse_approve_request(tx['data'])
     recipient_address = tx_data[0]