Defer processing backend error handling to parent

chaind/eth/chain.py

@@ -2,6 +2,7 @@
 from chainlib.interface import ChainInterface
 from chainlib.eth.block import (
         block_by_number,
+        block_latest,
         Block,
         )
 from chainlib.eth.tx import (
@@ -16,3 +17,4 @@ class EthChainInterface(ChainInterface):
         self._block_from_src = Block.from_src
         self._tx_receipt = receipt
         self._src_normalize = Tx.src_normalize
+        self._block_latest = block_latest

chaind/eth/runnable/queuer.py

@@ -6,6 +6,7 @@ import signal
 # external imports
 import chainlib.eth.cli
 import chaind.cli
+import chainqueue.cli
 from chaind.session import SessionController
 from chaind.setup import Environment
 from chaind.error import (
@@ -44,14 +45,18 @@ env = Environment(domain='eth', env=os.environ)
 arg_flags = chainlib.eth.cli.argflag_std_read
 argparser = chainlib.eth.cli.ArgumentParser(arg_flags)
 
+queue_arg_flags = 0
+chainqueue.cli.process_flags(argparser, queue_arg_flags)
+
 local_arg_flags = chaind.cli.argflag_local_base | chaind.cli.ChaindFlag.DISPATCH | chaind.cli.ChaindFlag.SOCKET
 chaind.cli.process_flags(argparser, local_arg_flags)
 
 args = argparser.parse_args()
 
-base_config_dir = [chaind.cli.config_dir]
+base_config_dir = [chainqueue.cli.config_dir, chaind.cli.config_dir]
 config = chainlib.eth.cli.Config.from_args(args, arg_flags, base_config_dir=base_config_dir)
 config = chaind.cli.process_config(config, args, local_arg_flags)
+config = chainqueue.cli.process_config(config, args, queue_arg_flags)
 config.add('eth', 'CHAIND_ENGINE', False)
 config.add('queue', 'CHAIND_COMPONENT', False)
 logg.debug('config loaded:\n{}'.format(config))
@@ -62,10 +67,18 @@ 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
+    adapter = ChaindFsAdapter(
+        settings.get('CHAIN_SPEC'),
+        settings.dir_for('queue'),
+        EthCacheTx,
+        dispatcher,
+        )
+    
+    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):
@@ -82,7 +95,7 @@ token_cache_store = CacheTokenTx(chain_spec, normalizer=tx_normalizer)
 dispatcher = EthDispatcher(conn)
 queue_adapter = ChaindFsAdapter(
         settings.get('CHAIN_SPEC'),
-        settings.get('SESSION_DATA_DIR'),
+        settings.dir_for('queue'),
         EthCacheTx,
         dispatcher,
         )

chaind/eth/runnable/send.py

@@ -10,6 +10,7 @@ import socket
 
 # external imports
 import chainlib.eth.cli
+import chainqueue.cli
 import chaind.cli
 from chaind.setup import Environment
 from chainlib.eth.gas import price
@@ -37,12 +38,15 @@ argparser.add_positional('source', required=False, type=str, help='Transaction s
 local_arg_flags = chaind.cli.argflag_local_socket_client | chaind.cli.ChaindFlag.TOKEN
 chaind.cli.process_flags(argparser, local_arg_flags)
 
+chainqueue.cli.process_flags(argparser, 0)
+
 args = argparser.parse_args()
 
 env = Environment(domain='eth', env=os.environ)
 
-base_config_dir = [chaind.cli.config_dir]
+base_config_dir = [chaind.cli.config_dir, chainqueue.cli.config_dir]
 config = chainlib.eth.cli.Config.from_args(args, arg_flags, base_config_dir=base_config_dir)
+config = chainqueue.cli.process_config(config, args, 0)
 config = chaind.cli.process_config(config, args, local_arg_flags)
 config.add(args.source, '_SOURCE', False)
 config.add('eth', 'CHAIND_ENGINE', False)

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
@@ -57,14 +56,9 @@ settings.process(config)
 
 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)
 

chaind/eth/token/base.py

@@ -9,7 +9,7 @@ logg = logging.getLogger(__name__)
 
 class BaseTokenResolver:
 
-    def __init__(self, chain_spec, sender, signer, gas_oracle, nonce_oracle):
+    def __init__(self, chain_spec, sender, signer, gas_oracle, nonce_oracle, advance_nonce=False):
         self.chain_spec = chain_spec
         self.chain_id = chain_spec.chain_id()
         self.signer = signer
@@ -19,6 +19,10 @@ class BaseTokenResolver:
         self.factory = None
         self.gas_limit_start = None
         self.gas_price_start = None
+        if advance_nonce:
+            self.nonce_getter = self.nonce_oracle.next_nonce
+        else:
+            self.nonce_getter = self.nonce_oracle.get_nonce
 
     
     def reset(self):
@@ -28,7 +32,7 @@ class BaseTokenResolver:
 
 
     def get_values(self, gas_value, value, executable_address=None):
-        nonce = self.nonce_oracle.get_nonce()
+        nonce = self.nonce_getter()
 
         if executable_address == None:
             return (value, 0, nonce)

chaind/eth/token/gas.py

@@ -9,7 +9,7 @@ from chaind.eth.token import BaseTokenResolver
 class GasTokenResolver(BaseTokenResolver):
 
     def __init__(self, chain_spec, sender, signer, gas_oracle, nonce_oracle):
-        super(GasTokenResolver, self).__init__(chain_spec, sender, signer, gas_oracle, nonce_oracle)
+        super(GasTokenResolver, self).__init__(chain_spec, sender, signer, gas_oracle, nonce_oracle, advance_nonce=True)
         self.factory = Gas(self.chain_spec, signer=self.signer, gas_oracle=self.gas_oracle, nonce_oracle=self.nonce_oracle)
 
 

requirements.txt

@@ -1,6 +1,6 @@
-chaind~=0.1.1
+chaind~=0.2.2
 hexathon~=0.1.5
-chainlib-eth~=0.1.0
+chainlib-eth~=0.1.1
 pyxdg~=0.27
 shep~=0.2.3
 funga-eth~=0.6.0

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = chaind-eth
-version = 0.1.3
+version = 0.2.1
 description = Queue server for ethereum
 author = Louis Holbrook
 author_email = dev@holbrook.no

setup.py

@@ -27,7 +27,7 @@ while True:
     l = f.readline()
     if l == '':
         break
-    test_requirements.append(l.rstrip())
+    erc20_requirements.append(l.rstrip())
 f.close()
 
 

tests/test_tx.py

@@ -11,7 +11,7 @@ from chainlib.chain import ChainSpec
 from chainqueue.cache import CacheTokenTx
 from chainlib.error import RPCException
 from chainlib.status import Status as TxStatus
-from chaind.unittest.common import TestChaindFsBase
+from chaind.unittest.fs import TestChaindFsBase
 from chaind.driver import QueueDriver
 from chaind.filter import StateFilter
 from chainlib.eth.gas import Gas