cic-internal-integration/apps/cic-eth/cic_eth/task.py

165 lines
4.6 KiB
Python
Raw Normal View History

2021-03-01 21:15:17 +01:00
# import
import time
import logging
import uuid
2021-03-01 21:15:17 +01:00
# external imports
import celery
import sqlalchemy
from chainlib.chain import ChainSpec
from chainlib.connection import RPCConnection
from chainlib.eth.constant import ZERO_ADDRESS
from chainlib.eth.nonce import RPCNonceOracle
from chainlib.eth.gas import RPCGasOracle
from cic_eth_registry import CICRegistry
from cic_eth_registry.error import UnknownContractError
2021-03-01 21:15:17 +01:00
2021-03-07 14:51:59 +01:00
# local imports
from cic_eth.error import SeppukuError
from cic_eth.db.models.base import SessionBase
from cic_eth.eth.util import CacheGasOracle
2021-03-07 14:51:59 +01:00
2021-05-31 17:34:16 +02:00
#logg = logging.getLogger().getChild(__name__)
logg = logging.getLogger()
2021-03-01 21:15:17 +01:00
celery_app = celery.current_app
2021-12-07 21:45:07 +01:00
class BaseTask(celery.Task):
session_func = SessionBase.create_session
call_address = ZERO_ADDRESS
2021-10-14 15:24:51 +02:00
trusted_addresses = []
min_fee_price = 1
2021-12-07 21:45:07 +01:00
min_fee_limit = 30000
2021-04-24 08:14:24 +02:00
default_token_address = None
default_token_symbol = None
default_token_name = None
default_token_decimals = None
run_dir = '/run'
def create_gas_oracle(self, conn, address=None, *args, **kwargs):
if address == None:
return RPCGasOracle(
conn,
2021-12-07 21:45:07 +01:00
code_callback=kwargs.get('code_callback', self.get_min_fee_limit),
min_price=self.min_fee_price,
id_generator=kwargs.get('id_generator'),
)
return CacheGasOracle(
conn,
address,
method=kwargs.get('method'),
min_price=self.min_fee_price,
id_generator=kwargs.get('id_generator'),
)
2021-12-07 21:45:07 +01:00
def get_min_fee_limit(self, code):
return self.min_fee_limit
def create_session(self):
return BaseTask.session_func()
def log_banner(self):
logg.debug('task {} root uuid {}'.format(self.__class__.__name__, self.request.root_id))
return
def on_failure(self, exc, task_id, args, kwargs, einfo):
if isinstance(exc, SeppukuError):
2021-09-01 18:54:10 +02:00
import liveness.linux
liveness.linux.reset(rundir=self.run_dir)
logg.critical(einfo)
msg = 'received critical exception {}, calling shutdown'.format(str(exc))
s = celery.signature(
'cic_eth.admin.ctrl.shutdown',
[msg],
queue=self.request.delivery_info.get('routing_key'),
)
s.apply_async()
class CriticalTask(BaseTask):
2021-03-01 21:15:17 +01:00
retry_jitter = True
retry_backoff = True
retry_backoff_max = 8
class CriticalSQLAlchemyTask(CriticalTask):
autoretry_for = (
sqlalchemy.exc.DatabaseError,
sqlalchemy.exc.TimeoutError,
sqlalchemy.exc.ResourceClosedError,
2021-03-01 21:15:17 +01:00
)
class CriticalWeb3Task(CriticalTask):
autoretry_for = (
2021-07-08 17:28:04 +02:00
ConnectionError,
2021-03-01 21:15:17 +01:00
)
safe_gas_threshold_amount = 60000 * 3
safe_gas_refill_amount = safe_gas_threshold_amount * 5
safe_gas_gifter_balance = safe_gas_threshold_amount * 5 * 100
2021-03-01 21:15:17 +01:00
class CriticalSQLAlchemyAndWeb3Task(CriticalWeb3Task):
2021-03-01 21:15:17 +01:00
autoretry_for = (
sqlalchemy.exc.DatabaseError,
sqlalchemy.exc.TimeoutError,
2021-07-08 17:28:04 +02:00
ConnectionError,
sqlalchemy.exc.ResourceClosedError,
2021-03-01 21:15:17 +01:00
)
2021-03-07 14:51:59 +01:00
class CriticalSQLAlchemyAndSignerTask(CriticalTask):
autoretry_for = (
sqlalchemy.exc.DatabaseError,
sqlalchemy.exc.TimeoutError,
sqlalchemy.exc.ResourceClosedError,
2021-03-07 14:51:59 +01:00
)
class CriticalWeb3AndSignerTask(CriticalWeb3Task):
2021-03-07 14:51:59 +01:00
autoretry_for = (
2021-07-08 17:28:04 +02:00
ConnectionError,
2021-03-07 14:51:59 +01:00
)
@celery_app.task()
def check_health(self):
pass
# TODO: registry / rpc methods should perhaps be moved to better named module
@celery_app.task()
def registry():
return CICRegistry.address
2021-05-31 17:34:16 +02:00
@celery_app.task(bind=True, base=BaseTask)
def registry_address_lookup(self, 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)
2021-05-31 17:34:16 +02:00
r = registry.by_address(address, sender_address=self.call_address)
return r
@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)
2021-05-31 17:34:16 +02:00
return registry.by_name(name, sender_address=self.call_address)
@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)