Syncer refactor

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

@@ -14,6 +14,7 @@ from cic_eth.db.enum import (
         StatusBits,
         dead,
         )
+from cic_eth.task import CriticalSQLAlchemyTask
 
 celery_app = celery.current_app
 
@@ -35,7 +36,7 @@ def __balance_outgoing_compatible(token_address, holder_address, chain_str):
     return delta
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def balance_outgoing(tokens, holder_address, chain_str):
     """Retrieve accumulated value of unprocessed transactions sent from the given address.
 
@@ -73,7 +74,7 @@ def __balance_incoming_compatible(token_address, receiver_address, chain_str):
     return delta
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def balance_incoming(tokens, receipient_address, chain_str):
     """Retrieve accumulated value of unprocessed transactions to be received by the given address.
 

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

@@ -10,6 +10,7 @@ from cic_registry.chain import ChainSpec
 from cic_eth.eth.rpc import RpcClient
 from cic_eth.db.models.otx import Otx
 from cic_eth.error import NotLocalTxError
+from cic_eth.task import CriticalSQLAlchemyAndWeb3Task
 
 celery_app = celery.current_app
 
@@ -17,7 +18,7 @@ logg = logging.getLogger()
 
 
 # TODO: This method does not belong in the _queue_ module, it operates across queue and network
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyAndWeb3Task)
 def tx_times(tx_hash, chain_str):
     chain_spec = ChainSpec.from_chain_str(chain_str)
     c = RpcClient(chain_spec)

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

@@ -26,6 +26,7 @@ from cic_eth.db.enum import (
         is_alive,
         dead,
         )
+from cic_eth.task import CriticalSQLAlchemyTask
 from cic_eth.eth.util import unpack_signed_raw_tx # TODO: should not be in same sub-path as package that imports queue.tx
 from cic_eth.error import NotLocalTxError
 from cic_eth.error import LockedError
@@ -86,7 +87,7 @@ def create(nonce, holder_address, tx_hash, signed_tx, chain_str, obsolete_predec
 
 
 # TODO: Replace set_* with single task for set status
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_sent_status(tx_hash, fail=False):
     """Used to set the status after a send attempt
 
@@ -118,7 +119,7 @@ def set_sent_status(tx_hash, fail=False):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_final_status(tx_hash, block=None, fail=False):
     """Used to set the status of an incoming transaction result. 
 
@@ -174,7 +175,7 @@ def set_final_status(tx_hash, block=None, fail=False):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_cancel(tx_hash, manual=False):
     """Used to set the status when a transaction is cancelled.
 
@@ -206,7 +207,7 @@ def set_cancel(tx_hash, manual=False):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_rejected(tx_hash):
     """Used to set the status when the node rejects sending a transaction to network
 
@@ -232,7 +233,7 @@ def set_rejected(tx_hash):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_fubar(tx_hash):
     """Used to set the status when an unexpected error occurs.
 
@@ -258,7 +259,7 @@ def set_fubar(tx_hash):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_manual(tx_hash):
     """Used to set the status when queue is manually changed
 
@@ -284,7 +285,7 @@ def set_manual(tx_hash):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_ready(tx_hash):
     """Used to mark a transaction as ready to be sent to network
 
@@ -310,7 +311,25 @@ def set_ready(tx_hash):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def set_dequeue(tx_hash):
+    session = SessionBase.create_session()
+    o = session.query(Otx).filter(Otx.tx_hash==tx_hash).first()
+    if o == None:
+        session.close()
+        raise NotLocalTxError('queue does not contain tx hash {}'.format(tx_hash))
+
+    session.flush()
+
+    o.dequeue(session=session)
+    session.commit()
+    session.close()
+
+    return tx_hash
+
+
+
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_waitforgas(tx_hash):
     """Used to set the status when a transaction must be deferred due to gas refill
 
@@ -336,7 +355,7 @@ def set_waitforgas(tx_hash):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_state_log(tx_hash):
 
     logs = []
@@ -355,7 +374,7 @@ def get_state_log(tx_hash):
     return logs
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_tx_cache(tx_hash):
     """Returns an aggregate dictionary of outgoing transaction data and metadata
 
@@ -404,7 +423,7 @@ def get_tx_cache(tx_hash):
     return tx
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_lock(address=None):
     """Retrieve all active locks
 
@@ -442,7 +461,7 @@ def get_lock(address=None):
     return locks
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_tx(tx_hash):
     """Retrieve a transaction queue record by transaction hash
 
@@ -453,7 +472,9 @@ def get_tx(tx_hash):
     :rtype: dict
     """
     session = SessionBase.create_session()
-    tx = session.query(Otx).filter(Otx.tx_hash==tx_hash).first()
+    q = session.query(Otx)
+    q = q.filter(Otx.tx_hash==tx_hash)
+    tx = q.first()
     if tx == None:
         session.close()
         raise NotLocalTxError('queue does not contain tx hash {}'.format(tx_hash))
@@ -469,7 +490,7 @@ def get_tx(tx_hash):
     return o
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_nonce_tx(nonce, sender, chain_id):
     """Retrieve all transactions for address with specified nonce
 
@@ -652,7 +673,7 @@ def get_upcoming_tx(status=StatusEnum.READYSEND, recipient=None, before=None, ch
     return txs
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_account_tx(address, as_sender=True, as_recipient=True, counterpart=None):
     """Returns all local queue transactions for a given Ethereum address