2021-02-01 18:12:51 +01:00
# standard imports
import os
import sys
import logging
import time
import argparse
import sys
import re
import datetime
# third-party imports
import confini
import celery
2021-03-22 22:10:52 +01:00
from cic_eth_registry import CICRegistry
from chainlib . chain import ChainSpec
2021-03-06 18:55:51 +01:00
from chainlib . eth . tx import unpack
2021-03-22 22:10:52 +01:00
from chainsyncer . error import SyncDone
2021-03-06 18:55:51 +01:00
from hexathon import strip_0x
2021-02-01 18:12:51 +01:00
# local imports
import cic_eth
from cic_eth . eth import RpcClient
from cic_eth . db import SessionBase
from cic_eth . db . enum import StatusEnum
2021-03-01 21:15:17 +01:00
from cic_eth . db . enum import StatusBits
2021-02-01 18:12:51 +01:00
from cic_eth . db . enum import LockEnum
from cic_eth . db import dsn_from_config
2021-03-01 21:15:17 +01:00
from cic_eth . queue . tx import (
get_upcoming_tx ,
set_dequeue ,
)
2021-02-01 18:12:51 +01:00
from cic_eth . admin . ctrl import lock_send
from cic_eth . eth . tx import send as task_tx_send
2021-03-01 21:15:17 +01:00
from cic_eth . error import (
PermanentTxError ,
TemporaryTxError ,
NotLocalTxError ,
)
2021-03-06 18:55:51 +01:00
#from cic_eth.eth.util import unpack_signed_raw_tx_hex
2021-02-01 18:12:51 +01:00
logging . basicConfig ( level = logging . WARNING )
logg = logging . getLogger ( )
logging . getLogger ( ' websockets.protocol ' ) . setLevel ( logging . CRITICAL )
logging . getLogger ( ' web3.RequestManager ' ) . setLevel ( logging . CRITICAL )
logging . getLogger ( ' web3.providers.WebsocketProvider ' ) . setLevel ( logging . CRITICAL )
logging . getLogger ( ' web3.providers.HTTPProvider ' ) . setLevel ( logging . CRITICAL )
config_dir = os . path . join ( ' /usr/local/etc/cic-eth ' )
argparser = argparse . ArgumentParser ( description = ' daemon that monitors transactions in new blocks ' )
2021-03-22 22:10:52 +01:00
argparser . add_argument ( ' -p ' , ' --provider ' , dest = ' p ' , type = str , help = ' rpc provider ' )
2021-02-01 18:12:51 +01:00
argparser . add_argument ( ' -c ' , type = str , default = config_dir , help = ' config root to use ' )
argparser . add_argument ( ' --env-prefix ' , default = os . environ . get ( ' CONFINI_ENV_PREFIX ' ) , dest = ' env_prefix ' , type = str , help = ' environment prefix for variables to overwrite configuration ' )
argparser . add_argument ( ' -q ' , type = str , default = ' cic-eth ' , help = ' celery queue to submit transaction tasks to ' )
argparser . add_argument ( ' -v ' , help = ' be verbose ' , action = ' store_true ' )
argparser . add_argument ( ' -vv ' , help = ' be more verbose ' , action = ' store_true ' )
args = argparser . parse_args ( sys . argv [ 1 : ] )
if args . v == True :
logging . getLogger ( ) . setLevel ( logging . INFO )
elif args . vv == True :
logging . getLogger ( ) . setLevel ( logging . DEBUG )
config_dir = os . path . join ( args . c )
os . makedirs ( config_dir , 0o777 , True )
config = confini . Config ( config_dir , args . env_prefix )
config . process ( )
# override args
config . censor ( ' PASSWORD ' , ' DATABASE ' )
config . censor ( ' PASSWORD ' , ' SSL ' )
logg . debug ( ' config loaded from {} : \n {} ' . format ( config_dir , config ) )
app = celery . Celery ( backend = config . get ( ' CELERY_RESULT_URL ' ) , broker = config . get ( ' CELERY_BROKER_URL ' ) )
queue = args . q
dsn = dsn_from_config ( config )
2021-02-12 03:27:29 +01:00
SessionBase . connect ( dsn , debug = config . true ( ' DATABASE_DEBUG ' ) )
2021-02-01 18:12:51 +01:00
2021-03-22 22:10:52 +01:00
chain_spec = ChainSpec . from_chain_str ( config . get ( ' CIC_CHAIN_SPEC ' ) )
2021-02-01 18:12:51 +01:00
2021-03-22 22:10:52 +01:00
RPCConnection . registry_location ( args . p , chain_spec , tag = ' default ' )
2021-02-01 18:12:51 +01:00
run = True
class DispatchSyncer :
2021-02-21 16:41:37 +01:00
yield_delay = 0.005
2021-02-01 18:12:51 +01:00
def __init__ ( self , chain_spec ) :
self . chain_spec = chain_spec
self . chain_id = chain_spec . chain_id ( )
def chain ( self ) :
return self . chain_spec
def process ( self , w3 , txs ) :
c = len ( txs . keys ( ) )
logg . debug ( ' processing {} txs {} ' . format ( c , list ( txs . keys ( ) ) ) )
chain_str = str ( self . chain_spec )
for k in txs . keys ( ) :
tx_raw = txs [ k ]
2021-03-06 18:55:51 +01:00
#tx = unpack_signed_raw_tx_hex(tx_raw, self.chain_spec.chain_id())
tx_raw_bytes = bytes . fromhex ( strip_0x ( tx_raw ) )
tx = unpack ( tx_raw_bytes , self . chain_spec . chain_id ( ) )
2021-03-01 21:15:17 +01:00
try :
set_dequeue ( tx [ ' hash ' ] )
except NotLocalTxError as e :
logg . warning ( ' dispatcher was triggered with non-local tx {} ' . format ( tx [ ' hash ' ] ) )
2021-03-06 18:55:51 +01:00
continue
2021-02-01 18:12:51 +01:00
s_check = celery . signature (
' cic_eth.admin.ctrl.check_lock ' ,
[
[ tx_raw ] ,
chain_str ,
LockEnum . QUEUE ,
tx [ ' from ' ] ,
] ,
queue = queue ,
)
s_send = celery . signature (
' cic_eth.eth.tx.send ' ,
[
chain_str ,
] ,
queue = queue ,
)
s_check . link ( s_send )
t = s_check . apply_async ( )
2021-03-01 21:15:17 +01:00
logg . info ( ' processed {} ' . format ( k ) )
2021-02-01 18:12:51 +01:00
def loop ( self , w3 , interval ) :
while run :
txs = { }
2021-03-01 21:15:17 +01:00
typ = StatusBits . QUEUED
2021-02-01 18:12:51 +01:00
utxs = get_upcoming_tx ( typ , chain_id = self . chain_id )
for k in utxs . keys ( ) :
txs [ k ] = utxs [ k ]
self . process ( w3 , txs )
2021-02-21 16:41:37 +01:00
if len ( utxs ) > 0 :
time . sleep ( self . yield_delay )
else :
time . sleep ( interval )
2021-02-01 18:12:51 +01:00
def main ( ) :
syncer = DispatchSyncer ( chain_spec )
try :
syncer . loop ( c . w3 , float ( config . get ( ' DISPATCHER_LOOP_INTERVAL ' ) ) )
2021-03-22 22:10:52 +01:00
except SyncDone as e :
2021-02-01 18:12:51 +01:00
sys . stderr . write ( " dispatcher done at block {} \n " . format ( e ) )
sys . exit ( 0 )
if __name__ == ' __main__ ' :
main ( )