Merge remote-tracking branch 'origin/master' into lash/celery-graphs

This commit is contained in:
nolash 2021-04-25 14:27:51 +02:00
commit e84c31f043
9 changed files with 170 additions and 78 deletions

View File

@ -60,6 +60,29 @@ class AdminApi:
self.call_address = call_address 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): def unlock(self, chain_spec, address, flags=None):
s_unlock = celery.signature( s_unlock = celery.signature(
'cic_eth.admin.ctrl.unlock', 'cic_eth.admin.ctrl.unlock',
@ -146,7 +169,6 @@ class AdminApi:
# TODO: This check should most likely be in resend task itself # TODO: This check should most likely be in resend task itself
tx_dict = s_get_tx_cache.apply_async().get() 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): if not is_alive(getattr(StatusEnum, tx_dict['status']).value):
raise TxStateChangeError('Cannot resend mined or obsoleted transaction'.format(txold_hash_hex)) raise TxStateChangeError('Cannot resend mined or obsoleted transaction'.format(txold_hash_hex))
@ -226,9 +248,6 @@ class AdminApi:
break break
last_nonce = nonce_otx last_nonce = nonce_otx
#nonce_cache = Nonce.get(address)
#nonce_w3 = self.w3.eth.getTransactionCount(address, 'pending')
return { return {
'nonce': { 'nonce': {
#'network': nonce_cache, #'network': nonce_cache,
@ -272,20 +291,6 @@ class AdminApi:
return s_nonce.apply_async() 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): 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. """Lists locally originated transactions for the given Ethereum address.
@ -348,6 +353,7 @@ class AdminApi:
# TODO: Add exception upon non-existent tx aswell as invalid tx data to docstring # 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): 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. """Output local and network details about a given transaction with local origin.
@ -370,7 +376,6 @@ class AdminApi:
if tx_raw != None: if tx_raw != None:
tx_hash = add_0x(keccak256_hex_to_hex(tx_raw)) tx_hash = add_0x(keccak256_hex_to_hex(tx_raw))
#tx_hash = self.w3.keccak(hexstr=tx_raw).hex()
s = celery.signature( s = celery.signature(
'cic_eth.queue.query.get_tx_cache', 'cic_eth.queue.query.get_tx_cache',
@ -386,38 +391,78 @@ class AdminApi:
source_token = None source_token = None
if tx['source_token'] != ZERO_ADDRESS: if tx['source_token'] != ZERO_ADDRESS:
try: if registry != None:
source_token = registry.by_address(tx['source_token']) try:
#source_token = CICRegistry.get_address(chain_spec, tx['source_token']).contract source_token = registry.by_address(tx['source_token'])
except UnknownContractError: except UnknownContractError:
#source_token_contract = self.w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=tx['source_token']) logg.warning('unknown source token contract {} (direct)'.format(tx['source_token']))
#source_token = CICRegistry.add_token(chain_spec, source_token_contract) else:
logg.warning('unknown source token contract {}'.format(tx['source_token'])) s = celery.signature(
'cic_eth.task.registry_address_lookup',
[
chain_spec.asdict(),
tx['source_token'],
],
queue=self.queue
)
t = s.apply_async()
source_token = t.get()
if source_token == None:
logg.warning('unknown source token contract {} (task pool)'.format(tx['source_token']))
destination_token = None destination_token = None
if tx['source_token'] != ZERO_ADDRESS: if tx['destination_token'] != ZERO_ADDRESS:
try: if registry != None:
#destination_token = CICRegistry.get_address(chain_spec, tx['destination_token']) try:
destination_token = registry.by_address(tx['destination_token']) destination_token = registry.by_address(tx['destination_token'])
except UnknownContractError: except UnknownContractError:
#destination_token_contract = self.w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=tx['source_token']) logg.warning('unknown destination token contract {}'.format(tx['destination_token']))
#destination_token = CICRegistry.add_token(chain_spec, destination_token_contract) else:
logg.warning('unknown destination token contract {}'.format(tx['destination_token'])) s = celery.signature(
'cic_eth.task.registry_address_lookup',
[
chain_spec.asdict(),
tx['destination_token'],
],
queue=self.queue
)
t = s.apply_async()
destination_token = t.get()
if destination_token == None:
logg.warning('unknown destination token contract {} (task pool)'.format(tx['destination_token']))
tx['sender_description'] = 'Custodial account' tx['sender_description'] = 'Custodial account'
tx['recipient_description'] = 'Custodial account' tx['recipient_description'] = 'Custodial account'
o = code(tx['sender']) 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: if len(strip_0x(r, allow_empty=True)) > 0:
try: if registry != None:
#sender_contract = CICRegistry.get_address(chain_spec, tx['sender']) try:
sender_contract = registry.by_address(tx['sender'], sender_address=self.call_address) sender_contract = registry.by_address(tx['sender'], sender_address=self.call_address)
tx['sender_description'] = 'Contract at {}'.format(tx['sender']) #sender_contract) tx['sender_description'] = 'Contract at {}'.format(tx['sender'])
except UnknownContractError: except UnknownContractError:
tx['sender_description'] = 'Unknown contract' tx['sender_description'] = 'Unknown contract'
except KeyError as e: except KeyError as e:
tx['sender_description'] = 'Unknown contract' 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: else:
s = celery.signature( s = celery.signature(
'cic_eth.eth.account.have', 'cic_eth.eth.account.have',
@ -446,16 +491,31 @@ class AdminApi:
tx['sender_description'] = role tx['sender_description'] = role
o = code(tx['recipient']) 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: if len(strip_0x(r, allow_empty=True)) > 0:
try: if registry != None:
#recipient_contract = CICRegistry.by_address(tx['recipient']) try:
recipient_contract = registry.by_address(tx['recipient']) recipient_contract = registry.by_address(tx['recipient'])
tx['recipient_description'] = 'Contract at {}'.format(tx['recipient']) #recipient_contract) tx['recipient_description'] = 'Contract at {}'.format(tx['recipient'])
except UnknownContractError as e: except UnknownContractError as e:
tx['recipient_description'] = 'Unknown contract' tx['recipient_description'] = 'Unknown contract'
except KeyError as e: except KeyError as e:
tx['recipient_description'] = 'Unknown contract' 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: else:
s = celery.signature( s = celery.signature(
'cic_eth.eth.account.have', 'cic_eth.eth.account.have',
@ -497,7 +557,8 @@ class AdminApi:
r = None r = None
try: try:
o = transaction(tx_hash) o = transaction(tx_hash)
r = self.rpc.do(o) t = self.proxy_do(chain_spec, o)
r = t.get()
if r != None: if r != None:
tx['network_status'] = 'Mempool' tx['network_status'] = 'Mempool'
except Exception as e: except Exception as e:
@ -506,7 +567,8 @@ class AdminApi:
if r != None: if r != None:
try: try:
o = receipt(tx_hash) 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)) logg.debug('h {} o {}'.format(tx_hash, o))
if int(strip_0x(r['status'])) == 1: if int(strip_0x(r['status'])) == 1:
tx['network_status'] = 'Confirmed' tx['network_status'] = 'Confirmed'
@ -521,11 +583,13 @@ class AdminApi:
pass pass
o = balance(tx['sender']) o = balance(tx['sender'])
r = self.rpc.do(o) t = self.proxy_do(chain_spec, o)
r = t.get()
tx['sender_gas_balance'] = r tx['sender_gas_balance'] = r
o = balance(tx['recipient']) 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['recipient_gas_balance'] = r
tx_unpacked = unpack(bytes.fromhex(strip_0x(tx['signed_tx'])), chain_spec) tx_unpacked = unpack(bytes.fromhex(strip_0x(tx['signed_tx'])), chain_spec)

View File

@ -29,5 +29,5 @@ def connect(rpc, chain_spec, registry_address):
CICRegistry.address = registry_address CICRegistry.address = registry_address
registry = CICRegistry(chain_spec, rpc) registry = CICRegistry(chain_spec, rpc)
registry_address = registry.by_name('ContractRegistry') registry_address = registry.by_name('ContractRegistry')
return registry return registry

View File

@ -23,7 +23,6 @@ default_config_dir = os.environ.get('CONFINI_DIR', '/usr/local/etc/cic')
argparser = argparse.ArgumentParser() argparser = argparse.ArgumentParser()
argparser.add_argument('-p', '--provider', dest='p', default='http://localhost:8545', type=str, help='Web3 provider url (http only)') argparser.add_argument('-p', '--provider', dest='p', default='http://localhost:8545', type=str, help='Web3 provider url (http only)')
argparser.add_argument('-r', '--registry-address', type=str, help='CIC registry address')
argparser.add_argument('-f', '--format', dest='f', default=default_format, type=str, help='Output format') argparser.add_argument('-f', '--format', dest='f', default=default_format, type=str, help='Output format')
argparser.add_argument('-c', type=str, default=default_config_dir, help='config root to use') argparser.add_argument('-c', type=str, default=default_config_dir, help='config root to use')
argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='chain spec') argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='chain spec')

View File

@ -36,7 +36,6 @@ from cic_eth.eth import (
from cic_eth.admin import ( from cic_eth.admin import (
debug, debug,
ctrl, ctrl,
token,
) )
from cic_eth.queue import ( from cic_eth.queue import (
query, query,

View File

@ -85,9 +85,6 @@ def main():
callback_queue=args.q, callback_queue=args.q,
) )
#register = not args.no_register
#logg.debug('register {}'.format(register))
#t = api.create_account(register=register)
t = api.transfer(config.get('_SENDER'), config.get('_RECIPIENT'), config.get('_VALUE'), config.get('_SYMBOL')) t = api.transfer(config.get('_SENDER'), config.get('_RECIPIENT'), config.get('_VALUE'), config.get('_SYMBOL'))
ps.get_message() ps.get_message()

View File

@ -81,10 +81,14 @@ chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
rpc = EthHTTPConnection(args.p) rpc = EthHTTPConnection(args.p)
registry_address = config.get('CIC_REGISTRY_ADDRESS') #registry_address = config.get('CIC_REGISTRY_ADDRESS')
admin_api = AdminApi(rpc) admin_api = AdminApi(rpc)
t = admin_api.registry()
registry_address = t.get()
logg.info('got registry address from task pool: {}'.format(registry_address))
trusted_addresses_src = config.get('CIC_TRUST_ADDRESS') trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
if trusted_addresses_src == None: if trusted_addresses_src == None:
logg.critical('At least one trusted address must be declared in CIC_TRUST_ADDRESS') logg.critical('At least one trusted address must be declared in CIC_TRUST_ADDRESS')
@ -151,14 +155,16 @@ def main():
txs = [] txs = []
renderer = render_tx renderer = render_tx
if len(config.get('_QUERY')) > 66: if len(config.get('_QUERY')) > 66:
registry = connect_registry(rpc, chain_spec, registry_address) #registry = connect_registry(rpc, chain_spec, registry_address)
admin_api.tx(chain_spec, tx_raw=config.get('_QUERY'), registry=registry, renderer=renderer) #admin_api.tx(chain_spec, tx_raw=config.get('_QUERY'), registry=registry, renderer=renderer)
admin_api.tx(chain_spec, tx_raw=config.get('_QUERY'), renderer=renderer)
elif len(config.get('_QUERY')) > 42: elif len(config.get('_QUERY')) > 42:
registry = connect_registry(rpc, chain_spec, registry_address) #registry = connect_registry(rpc, chain_spec, registry_address)
admin_api.tx(chain_spec, tx_hash=config.get('_QUERY'), registry=registry, renderer=renderer) #admin_api.tx(chain_spec, tx_hash=config.get('_QUERY'), registry=registry, renderer=renderer)
admin_api.tx(chain_spec, tx_hash=config.get('_QUERY'), renderer=renderer)
elif len(config.get('_QUERY')) == 42: elif len(config.get('_QUERY')) == 42:
registry = connect_registry(rpc, chain_spec, registry_address) #registry = connect_registry(rpc, chain_spec, registry_address)
txs = admin_api.account(chain_spec, config.get('_QUERY'), include_recipient=False, renderer=render_account) txs = admin_api.account(chain_spec, config.get('_QUERY'), include_recipient=False, renderer=render_account)
renderer = render_account renderer = render_account
elif len(config.get('_QUERY')) >= 4 and config.get('_QUERY')[:4] == 'lock': elif len(config.get('_QUERY')) >= 4 and config.get('_QUERY')[:4] == 'lock':

View File

@ -7,9 +7,13 @@ import uuid
# external imports # external imports
import celery import celery
import sqlalchemy import sqlalchemy
from chainlib.chain import ChainSpec
from chainlib.connection import RPCConnection
from chainlib.eth.constant import ZERO_ADDRESS from chainlib.eth.constant import ZERO_ADDRESS
from chainlib.eth.nonce import RPCNonceOracle from chainlib.eth.nonce import RPCNonceOracle
from chainlib.eth.gas import RPCGasOracle from chainlib.eth.gas import RPCGasOracle
from cic_eth_registry import CICRegistry
from cic_eth_registry.error import UnknownContractError
import liveness.linux import liveness.linux
# local imports # local imports
@ -101,12 +105,35 @@ class CriticalWeb3AndSignerTask(CriticalTask):
safe_gas_refill_amount = safe_gas_threshold_amount * 5 safe_gas_refill_amount = safe_gas_threshold_amount * 5
@celery_app.task(bind=True, base=BaseTask)
def hello(self):
time.sleep(0.1)
return id(SessionBase.create_session)
@celery_app.task() @celery_app.task()
def check_health(self): def check_health(self):
pass pass
# TODO: registry / rpc methods should perhaps be moved to better named module
@celery_app.task()
def registry():
return CICRegistry.address
@celery_app.task()
def registry_address_lookup(chain_spec_dict, address, connection_tag='default'):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
conn = RPCConnection.connect(chain_spec, tag=connection_tag)
registry = CICRegistry(chain_spec, conn)
return registry.by_address(address)
@celery_app.task(throws=(UnknownContractError,))
def registry_name_lookup(chain_spec_dict, name, connection_tag='default'):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
conn = RPCConnection.connect(chain_spec, tag=connection_tag)
registry = CICRegistry(chain_spec, conn)
return registry.by_name(name)
@celery_app.task()
def rpc_proxy(chain_spec_dict, o, connection_tag='default'):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
conn = RPCConnection.connect(chain_spec, tag=connection_tag)
return conn.do(o)

View File

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

View File

@ -1,5 +1,5 @@
cic-base[full_graph]==0.1.2a77 cic-base[full_graph]==0.1.2b2
sarafu-faucet==0.0.2a28 sarafu-faucet==0.0.2a28
cic-eth==0.11.0b6 cic-eth==0.11.0b10
cic-types==0.1.0a10 cic-types==0.1.0a10
crypto-dev-signer==0.4.14b2 crypto-dev-signer==0.4.14b3