Add tracker tx filter test

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:

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

@@ -387,7 +387,7 @@ def resend_with_higher_gas(self, txold_hash_hex, chain_spec_dict, gas=None, defa
         r = rpc.do(o)
         current_gas_price = int(r, 16)
         if tx['gasPrice'] > current_gas_price:
-            logg.info('Network gas price {} is lower than overdue tx gas price {}'.format(curent_gas_price, tx['gasPrice']))
+            logg.info('Network gas price {} is lower than overdue tx gas price {}'.format(current_gas_price, tx['gasPrice']))
             #tx['gasPrice'] = int(tx['gasPrice'] * default_factor)
             new_gas_price = tx['gasPrice'] + 1
         else:

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

@@ -98,3 +98,12 @@ def get_state_log(chain_spec_dict, tx_hash):
     r = chainqueue.state.get_state_log(chain_spec, tx_hash, session=session)
     session.close()
     return r
+
+
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def obsolete(chain_spec_dict, tx_hash, final):
+    chain_spec = ChainSpec.from_dict(chain_spec_dict)
+    session = SessionBase.create_session()
+    r = chainqueue.state.obsolete_by_cache(chain_spec, tx_hash, final, session=session)
+    session.close()
+    return r

apps/cic-eth/cic_eth/runnable/daemons/filters/tx.py

@@ -33,7 +33,7 @@ class TxFilter(SyncFilter):
         logg.info('tx filter match on {}'.format(otx.tx_hash))
         db_session.flush()
         SessionBase.release_session(db_session)
-        s = celery.signature(
+        s_final_state = celery.signature(
                 'cic_eth.queue.state.set_final',
                 [
                     self.chain_spec.asdict(),
@@ -43,7 +43,18 @@ class TxFilter(SyncFilter):
                     ],
                 queue=self.queue,
                 )
-        t = s.apply_async()
+        s_obsolete_state = celery.signature(
+                'cic_eth.queue.state.obsolete',
+                [
+                    self.chain_spec.asdict(),
+                    add_0x(tx_hash_hex),
+                    True,
+                    ],
+                queue=self.queue,
+                )
+        #s_final_state.link(s_obsolete_state)
+        t = celery.group(s_obsolete_state, s_final_state)()
+
         return t
 
 

apps/cic-eth/cic_eth/runnable/daemons/retry.py

@@ -21,21 +21,22 @@ from chainlib.eth.block import (
 from chainsyncer.driver import HeadSyncer
 from chainsyncer.backend import MemBackend
 from chainsyncer.error import NoBlockForYou
+from chainqueue.db.enum import (
+        StatusEnum,
+        StatusBits,
+        )
+from chainqueue.state import obsolete_by_cache
 
 # local imports
 from cic_eth.db import dsn_from_config
 from cic_eth.db import SessionBase
-from cic_eth.queue.tx import (
+from cic_eth.queue.query import (
         get_status_tx,
         get_tx,
 #        get_upcoming_tx,
         )
 from cic_eth.admin.ctrl import lock_send
-from cic_eth.db.enum import (
+from cic_eth.db.enum import LockEnum
-        StatusEnum,
-        StatusBits,
-        LockEnum,
-        )
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -203,6 +204,7 @@ class StragglerFilter:
 
     def filter(self, conn, block, tx, db_session=None):
         logg.debug('tx {}'.format(tx))
+        obsolete_by_cache(self.chain_spec, tx.hash, False, session=db_session)
         s_send = celery.signature(
                 'cic_eth.eth.gas.resend_with_higher_gas',
                 [
@@ -248,12 +250,16 @@ class RetrySyncer(HeadSyncer):
 
     def process(self, conn, block):
         before = datetime.datetime.utcnow() - datetime.timedelta(seconds=self.stalled_grace_seconds)
+        session = SessionBase.create_session()
         stalled_txs = get_status_tx(
+                self.chain_spec,
                 StatusBits.IN_NETWORK.value,
                 not_status=StatusBits.FINAL | StatusBits.MANUAL | StatusBits.OBSOLETE,
                 before=before,
                 limit=self.batch_size,
+                session=session,
                 )
+        session.close()
 #        stalled_txs = get_upcoming_tx(
 #                status=StatusBits.IN_NETWORK.value, 
 #                not_status=StatusBits.FINAL | StatusBits.MANUAL | StatusBits.OBSOLETE,

apps/cic-eth/requirements.txt

@@ -1,4 +1,4 @@
-cic-base~=0.1.2a58
+cic-base~=0.1.2a59
 celery==4.4.7
 crypto-dev-signer~=0.4.14a17
 confini~=0.3.6rc3
@@ -16,7 +16,7 @@ semver==2.13.0
 websocket-client==0.57.0
 moolb~=0.1.1b2
 eth-address-index~=0.1.1a7
-chainlib~=0.0.2a2
+chainlib~=0.0.2a3
 hexathon~=0.0.1a7
 chainsyncer~=0.0.1a21
 chainqueue~=0.0.1a3

apps/cic-eth/tests/filters/tx.py

@@ -0,0 +1,110 @@
+# external imports
+from chainlib.connection import RPCConnection
+from chainlib.eth.nonce import OverrideNonceOracle
+from chainlib.eth.tx import (
+        TxFormat,
+        unpack,
+        Tx,
+        )
+from chainlib.eth.gas import (
+        Gas,
+        OverrideGasOracle,
+        )
+from chainlib.eth.block import (
+        block_latest,
+        block_by_number,
+        Block,
+        )
+from chainqueue.db.models.otx import Otx
+from chainqueue.db.enum import StatusBits
+from chainqueue.tx import create as queue_create
+from chainqueue.state import (
+        set_reserved,
+        set_ready,
+        set_sent,
+        )
+
+from hexathon import strip_0x
+
+# local imports
+from cic_eth.runnable.daemons.filters.tx import TxFilter
+from cic_eth.eth.gas import cache_gas_data
+
+
+def test_tx(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        agent_roles,
+        celery_worker,
+        ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = OverrideNonceOracle(agent_roles['ALICE'], 42)
+    gas_oracle = OverrideGasOracle(price=1000000000, limit=21000)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+    queue_create(
+            default_chain_spec,
+            42,
+            agent_roles['ALICE'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+
+    set_ready(default_chain_spec, tx_hash_hex, session=init_database)
+    set_reserved(default_chain_spec, tx_hash_hex, session=init_database)
+    set_sent(default_chain_spec, tx_hash_hex, session=init_database)
+    tx_hash_hex_orig = tx_hash_hex
+
+    gas_oracle = OverrideGasOracle(price=1100000000, limit=21000)
+    (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
+    queue_create(
+            default_chain_spec,
+            42,
+            agent_roles['ALICE'],
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            session=init_database,
+            )
+    cache_gas_data(
+            tx_hash_hex,
+            tx_signed_raw_hex,
+            default_chain_spec.asdict(),
+            )
+
+    set_ready(default_chain_spec, tx_hash_hex, session=init_database)
+    set_reserved(default_chain_spec, tx_hash_hex, session=init_database)
+    set_sent(default_chain_spec, tx_hash_hex, session=init_database)
+
+    fltr = TxFilter(default_chain_spec, None)
+
+    o = block_latest()
+    r = eth_rpc.do(o)
+    o = block_by_number(r, include_tx=False)
+    r = eth_rpc.do(o)
+    block = Block(r)
+    block.txs = [tx_hash_hex]
+
+    tx_signed_raw_bytes = bytes.fromhex(strip_0x(tx_signed_raw_hex))
+    tx_src = unpack(tx_signed_raw_bytes, default_chain_spec)
+    tx = Tx(tx_src, block=block)
+    t = fltr.filter(eth_rpc, block, tx, db_session=init_database)
+
+    t.get()
+    assert t.successful()
+
+    otx = Otx.load(tx_hash_hex_orig, session=init_database)
+    assert otx.status & StatusBits.OBSOLETE == StatusBits.OBSOLETE
+    assert otx.status & StatusBits.FINAL == StatusBits.FINAL
+
+    otx = Otx.load(tx_hash_hex, session=init_database)
+    assert otx.status & StatusBits.OBSOLETE == 0
+    assert otx.status & StatusBits.FINAL == StatusBits.FINAL

apps/contract-migration/docker/Dockerfile

@@ -57,7 +57,7 @@ WORKDIR /home/grassroots
 USER grassroots
 
 ARG pip_extra_index_url=https://pip.grassrootseconomics.net:8433
-ARG cic_base_version=0.1.2a58
+ARG cic_base_version=0.1.2a59
 ARG cic_eth_version=0.11.0a4
 ARG sarafu_faucet_version=0.0.2a16
 ARG cic_contracts_version=0.0.2a2