chaind-eth

commit 9522729fa04e9252987238f2a051bd5e059b5522
parent 097a6f4b532eac0f22a4734c0e166573c28d6279
Author: lash <dev@holbrook.no>
Date:   Sat, 30 Apr 2022 18:44:52 +0000

Catch backend race

Diffstat:
M
chaind/eth/runnable/queuer.py
 | 
24
++++++++++++++++++++----
M
chaind/eth/runnable/syncer.py
 | 
9
+--------
M
requirements.txt
 | 
2
+-
M
setup.cfg
 | 
2
+-

4 files changed, 23 insertions(+), 14 deletions(-)
diff --git a/chaind/eth/runnable/queuer.py b/chaind/eth/runnable/queuer.py
@@ -67,10 +67,26 @@ settings.process(config)
 logg.debug('settings:\n{}'.format(settings))
 
 
-def process_outgoing(chain_spec, adapter, rpc, limit=100):
-    upcoming = adapter.upcoming()
-    logg.info('process {} {} {}'.format(chain_spec, adapter, rpc))
-    logg.info('upcoming {}'.format(upcoming))
+def process_outgoing(chain_spec, adapter, rpc, limit=50):
+    adapter = None
+    process_err = None
+    for i in range(2):
+        try:
+            adapter = ChaindFsAdapter(
+                settings.get('CHAIN_SPEC'),
+                settings.dir_for('queue'),
+                EthCacheTx,
+                dispatcher,
+                )
+        except BackendIntegrityError as e:
+            process_err = e
+            continue
+
+    if adapter == None:
+        raise BackendIntegrityError(process_err)
+    
+    upcoming = adapter.upcoming(limit=limit)
+    logg.info('processor has {} candidates for {}, processing with limit {} adapter {} rpc {}'.format(len(upcoming), chain_spec, limit, adapter, rpc))
     i = 0
     for tx_hash in upcoming:
         if adapter.dispatch(tx_hash):
diff --git a/chaind/eth/runnable/syncer.py b/chaind/eth/runnable/syncer.py
@@ -8,7 +8,6 @@ import chainsyncer.cli
 import chaind.cli
 from chaind.setup import Environment
 from chaind.filter import StateFilter
-from chaind.adapters.fs import ChaindFsAdapter
 from chainlib.eth.block import block_latest
 from hexathon import strip_0x
 from chainsyncer.store.fs import SyncFsStore
@@ -59,13 +58,7 @@ logg.debug('settings:\n{}'.format(settings))
 
 
 def main():
-    queue_adapter = ChaindFsAdapter(
-        settings.get('CHAIN_SPEC'),
-        settings.dir_for('queue'),
-        EthCacheTx,
-        None,
-        )
-    fltr = StateFilter(queue_adapter)
+    fltr = StateFilter(settings.get('CHAIN_SPEC'), settings.dir_for('queue'), EthCacheTx)
     sync_store = SyncFsStore(settings.get('SESSION_DATA_DIR'), session_id=settings.get('SESSION_ID'))
     sync_store.register(fltr)
 
diff --git a/requirements.txt b/requirements.txt
@@ -1,4 +1,4 @@
-chaind~=0.1.3
+chaind~=0.2.0
 hexathon~=0.1.5
 chainlib-eth~=0.1.1
 pyxdg~=0.27
diff --git a/setup.cfg b/setup.cfg
@@ -1,6 +1,6 @@
 [metadata]
 name = chaind-eth
-version = 0.1.5
+version = 0.2.0
 description = Queue server for ethereum
 author = Louis Holbrook
 author_email = dev@holbrook.no