Add db session creator to retrier backend

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

@@ -0,0 +1,34 @@
+# standard imports
+import logging
+
+# external imports
+import celery
+from chainqueue.state import obsolete_by_cache
+
+logg = logging.getLogger()
+
+
+
+class StragglerFilter:
+
+    def __init__(self, chain_spec, queue='cic-eth'):
+        self.chain_spec = chain_spec
+        self.queue = queue
+
+
+    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',
+                [
+                    tx.hash,
+                    self.chain_spec.asdict(),
+                ],
+                queue=self.queue,
+        )
+        return s_send.apply_async()
+
+
+    def __str__(self):
+        return 'stragglerfilter'

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

@@ -4,39 +4,22 @@ import sys
 import logging
 import argparse
 import re
-import datetime
 
 # external imports
 import confini
 import celery
 from cic_eth_registry import CICRegistry
 from chainlib.chain import ChainSpec
-from chainlib.eth.tx import unpack
 from chainlib.connection import RPCConnection
-from chainlib.eth.block import (
+from chainsyncer.filter import SyncFilter
-        block_latest,
-        block_by_number,
-        Block,
-        )
-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.query import (
-        get_status_tx,
-        get_tx,
-#        get_upcoming_tx,
-        )
 from cic_eth.admin.ctrl import lock_send
 from cic_eth.db.enum import LockEnum
+from cic_eth.runnable.daemons.filters.straggler import StragglerFilter
+from cic_eth.sync.retry import RetrySyncer
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -90,194 +73,55 @@ SessionBase.connect(dsn, debug=config.true('DATABASE_DEBUG'))
 
 straggler_delay = int(config.get('CIC_TX_RETRY_DELAY'))
 
-# TODO: we already have the signed raw tx in get, so its a waste of cycles to get_tx here
+## TODO: we already have the signed raw tx in get, so its a waste of cycles to get_tx here
-def sendfail_filter(w3, tx_hash, rcpt, chain_spec):
+#def sendfail_filter(w3, tx_hash, rcpt, chain_spec):
-    tx_dict = get_tx(tx_hash)
+#    tx_dict = get_tx(tx_hash)
-    tx = unpack(tx_dict['signed_tx'], chain_spec)
+#    tx = unpack(tx_dict['signed_tx'], chain_spec)
-    logg.debug('submitting tx {} for retry'.format(tx_hash))
+#    logg.debug('submitting tx {} for retry'.format(tx_hash))
-    s_check = celery.signature(
+#    s_check = celery.signature(
-            'cic_eth.admin.ctrl.check_lock',
+#            'cic_eth.admin.ctrl.check_lock',
-            [
+#            [
-                tx_hash,
+#                tx_hash,
-                chain_str,
+#                chain_str,
-                LockEnum.QUEUE,
+#                LockEnum.QUEUE,
-                tx['from'],
+#                tx['from'],
-                ],
+#                ],
-            queue=queue,
+#            queue=queue,
-            )
+#            )
-#    s_resume = celery.signature(
+##    s_resume = celery.signature(
-#            'cic_eth.eth.tx.resume_tx',
+##            'cic_eth.eth.tx.resume_tx',
+##            [
+##                chain_str,
+##                ],
+##            queue=queue,
+##            )
+#    
+##    s_retry_status = celery.signature(
+##            'cic_eth.queue.state.set_ready',
+##            [],
+##            queue=queue,
+##    )
+#    s_resend = celery.signature(
+#            'cic_eth.eth.gas.resend_with_higher_gas',
 #            [
 #                chain_str,
 #                ],
 #            queue=queue,
 #            )
-    
-#    s_retry_status = celery.signature(
-#            'cic_eth.queue.state.set_ready',
-#            [],
-#            queue=queue,
-#    )
-    s_resend = celery.signature(
-            'cic_eth.eth.gas.resend_with_higher_gas',
-            [
-                chain_str,
-                ],
-            queue=queue,
-            )
-
-    #s_resume.link(s_retry_status)
-    #s_check.link(s_resume)
-    s_check.link(s_resend)
-    s_check.apply_async()
-
-
-# TODO: can we merely use the dispatcher instead?
-def dispatch(conn, chain_spec):
-    txs = get_status_tx(StatusEnum.RETRY, before=datetime.datetime.utcnow())
-    if len(txs) == 0:
-        logg.debug('no retry state txs found')
-        return
-    #signed_txs = list(txs.values())
-    #logg.debug('signed txs {} chain {}'.format(signed_txs, chain_str))
-    #for tx in signed_txs:
-    for k in txs.keys():
-        #tx_cache = get_tx_cache(k)
-        tx_raw = txs[k]
-        tx = unpack(tx_raw, chain_spec)
-
-        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()
-
-#        try:
-#            r = t.get()
-#            logg.debug('submitted as {} result {} with queue task {}'.format(t, r, t.children[0].get()))
-#        except PermanentTxError as e:
-#            logg.error('tx {} permanently failed: {}'.format(tx, e))
-#        except TemporaryTxError as e:
-#            logg.error('tx {} temporarily failed: {}'.format(tx, e))
-
 #
-#
+#    #s_resume.link(s_retry_status)
-#def straggler_filter(w3, tx, rcpt, chain_str):
+#    #s_check.link(s_resume)
-#    before = datetime.datetime.utcnow() - datetime.timedelta(seconds=straggler_delay)
+#    s_check.link(s_resend)
-#    txs = get_status_tx(StatusEnum.SENT, before)
+#    s_check.apply_async()
-#    if len(txs) == 0:
-#        logg.debug('no straggler txs found')
-#        return
-#    txs = list(txs.keys())
-#    logg.debug('straggler txs {} chain {}'.format(signed_txs, chain_str))
-#    s_send = celery.signature(
-#            'cic_eth.eth.gas.resend_with_higher_gas',
-#            [
-#                txs,
-#                chain_str,
-#            ],
-#            queue=queue,
-#    )
-#    s_send.apply_async()
 
 
-class StragglerFilter:
-
-    def __init__(self, chain_spec, queue='cic-eth'):
-        self.chain_spec = chain_spec
-        self.queue = queue
-
-
-    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',
-                [
-                    tx,
-                    self.chain_spec.asdict(),
-                ],
-                queue=self.queue,
-        )
-        return s_send.apply_async()
-        #return s_send
-
-
-    def __str__(self):
-        return 'stragglerfilter'
-
-
-class RetrySyncer(HeadSyncer):
-
-    def __init__(self, conn, chain_spec, stalled_grace_seconds, batch_size=50, failed_grace_seconds=None):
-        backend = MemBackend(chain_spec, None)
-        super(RetrySyncer, self).__init__(backend)
-        self.chain_spec = chain_spec
-        if failed_grace_seconds == None:
-            failed_grace_seconds = stalled_grace_seconds
-        self.stalled_grace_seconds = stalled_grace_seconds
-        self.failed_grace_seconds = failed_grace_seconds
-        self.batch_size = batch_size
-        self.conn = conn
-
-
-    def get(self, conn):
-        o = block_latest()
-        r = conn.do(o)
-        (pair, flags) = self.backend.get()
-        n = int(r, 16)
-        if n == pair[0]:
-            raise NoBlockForYou('block {} already checked'.format(n))
-        o = block_by_number(n)
-        r = conn.do(o)
-        b = Block(r)
-        return b
-
-
-    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,
-#                before=before,
-#                limit=self.batch_size,
-#                )
-        for tx in stalled_txs:
-            self.filter.apply(self.conn, block, tx)
-        self.backend.set(block.number, 0)
-
 
 def main(): 
-    #o = block_latest()
     conn = RPCConnection.connect(chain_spec, 'default')
-    #block = conn.do(o)
     syncer = RetrySyncer(conn, chain_spec, straggler_delay, batch_size=config.get('_BATCH_SIZE'))
     syncer.backend.set(0, 0)
-    syncer.add_filter(StragglerFilter(chain_spec, queue=queue))
+    fltr = StragglerFilter(chain_spec, queue=queue)
+    syncer.add_filter(fltr)
     syncer.loop(float(straggler_delay), conn)
 
 

apps/cic-eth/cic_eth/sync/retry.py

@@ -0,0 +1,92 @@
+# standard imports
+import logging
+import datetime
+
+# external imports
+from chainsyncer.driver import HeadSyncer
+from chainsyncer.backend import MemBackend
+from chainsyncer.error import NoBlockForYou
+from chainlib.eth.block import (
+        block_by_number,
+        block_latest,
+        Block,
+        )
+from chainlib.eth.tx import (
+        unpack,
+        Tx,
+        )
+from cic_eth.queue.query import get_status_tx
+from chainqueue.db.enum import StatusBits
+from hexathon import strip_0x
+
+# local imports
+from cic_eth.db import SessionBase
+
+logg = logging.getLogger()
+
+
+class DbSessionMemBackend(MemBackend):
+
+    def connect(self):
+        self.db_session = SessionBase.create_session()
+        return self.db_session
+
+
+    def disconnect(self):
+        self.db_session.close()
+        self.db_session = None
+
+
+class RetrySyncer(HeadSyncer):
+
+    def __init__(self, conn, chain_spec, stalled_grace_seconds, batch_size=50, failed_grace_seconds=None):
+        backend = DbSessionMemBackend(chain_spec, None)
+        super(RetrySyncer, self).__init__(backend)
+        self.chain_spec = chain_spec
+        if failed_grace_seconds == None:
+            failed_grace_seconds = stalled_grace_seconds
+        self.stalled_grace_seconds = stalled_grace_seconds
+        self.failed_grace_seconds = failed_grace_seconds
+        self.batch_size = batch_size
+        self.conn = conn
+
+
+    def get(self, conn):
+        o = block_latest()
+        r = conn.do(o)
+        (pair, flags) = self.backend.get()
+        n = int(r, 16)
+        if n == pair[0]:
+            raise NoBlockForYou('block {} already checked'.format(n))
+        o = block_by_number(n)
+        r = conn.do(o)
+        b = Block(r)
+        return b
+
+
+    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,
+#                before=before,
+#                limit=self.batch_size,
+#                )
+        for tx_signed_raw_hex in stalled_txs.values():
+            tx_signed_raw_bytes = bytes.fromhex(strip_0x(tx_signed_raw_hex))
+            tx_src = unpack(tx_signed_raw_bytes, self.chain_spec)
+            tx = Tx(tx_src)
+            self.filter.apply(self.conn, block, tx)
+        self.backend.set(block.number, 0)
+
+

apps/cic-eth/docker/Dockerfile

@@ -29,7 +29,7 @@ RUN /usr/local/bin/python -m pip install --upgrade pip
 #	python merge_requirements.py | tee merged_requirements.txt 
 #RUN cd cic-base && \
 #	pip install $pip_extra_index_url_flag -r ./merged_requirements.txt 
-RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a58
+RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a60
 
 COPY cic-eth/scripts/ scripts/
 COPY cic-eth/setup.cfg cic-eth/setup.py ./

apps/cic-eth/requirements.txt

@@ -1,4 +1,4 @@
-cic-base~=0.1.2a59
+cic-base~=0.1.2a60
 celery==4.4.7
 crypto-dev-signer~=0.4.14a17
 confini~=0.3.6rc3
@@ -16,10 +16,10 @@ semver==2.13.0
 websocket-client==0.57.0
 moolb~=0.1.1b2
 eth-address-index~=0.1.1a7
-chainlib~=0.0.2a3
+chainlib~=0.0.2a4
 hexathon~=0.0.1a7
 chainsyncer~=0.0.1a21
-chainqueue~=0.0.1a3
+chainqueue~=0.0.1a5
 pysha3==1.0.2
 coincurve==15.0.0
 sarafu-faucet~=0.0.2a16

apps/cic-eth/setup.cfg

@@ -37,6 +37,7 @@ packages =
 	cic_eth.runnable.daemons
 	cic_eth.runnable.daemons.filters
 	cic_eth.callbacks
+	cic_eth.sync
 scripts =
 	./scripts/migrate.py
 

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

@@ -0,0 +1,89 @@
+# 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.straggler import StragglerFilter
+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)
+
+    fltr = StragglerFilter(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(None, block, tx, db_session=init_database)
+    tx_hash_hex_successor = t.get_leaf()
+
+    assert t.successful()
+    assert tx_hash_hex_successor != tx_hash_hex
+
+    otx = Otx.load(tx_hash_hex, session=init_database)
+    assert otx.status & StatusBits.OBSOLETE > 0
+    assert otx.status & (StatusBits.FINAL | StatusBits.QUEUED | StatusBits.RESERVED) == 0
+
+    otx = Otx.load(tx_hash_hex_successor, session=init_database)
+    assert otx.status == StatusBits.QUEUED

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.2a59
+ARG cic_base_version=0.1.2a60
 ARG cic_eth_version=0.11.0a4
 ARG sarafu_faucet_version=0.0.2a16
 ARG cic_contracts_version=0.0.2a2

docker-compose.yml

@@ -358,7 +358,7 @@ services:
       CELERY_BROKER_URL: ${CELERY_BROKER_URL:-redis://redis}
       CELERY_RESULT_URL: ${CELERY_RESULT_URL:-redis://redis}
       TASKS_TRANSFER_CALLBACKS: $TASKS_TRANSFER_CALLBACKS
-      CIC_TX_RETRY_DELAY: 15
+      CIC_TX_RETRY_DELAY: 60
       BATCH_SIZE: ${RETRIER_BATCH_SIZE:-50}
         #DATABASE_DEBUG: 1
     depends_on: