Update changelog

CHANGELOG

@@ -1,3 +1,35 @@
+- 0.1.15
+	* Upgrade shep to avoid sync in persist set
+- 0.1.14
+	* Upgrade shep to handle exception in filestore list
+- 0.1.13
+	* Remove sync on each get
+	* Upgrade shep to guarantee atomic state lock state
+- 0.1.12
+	* Raise correct exception from index store exists check
+- 0.1.11
+	* Allow for sync skip in store instantiation
+- 0.1.10
+	* Improve logging
+- 0.1.9
+	* Upgrade deps
+- 0.1.8
+	* Upgrade deps
+- 0.1.7 
+	* Improve logging
+- 0.1.6
+	* Sort upcoming queue item chronologically
+	* Add unit testing for upcoming query method
+- 0.1.5
+	* Add reserved state check method
+- 0.1.4
+	* Dependency cleanups
+- 0.1.3
+	* Add CLI args and config handling, settings object
+- 0.1.2
+	* Add CLI inspection tools
+- 0.1.1
+	* 
 - 0.1.0
 	* Replace state transitions with shep
 - 0.0.3

chainqueue/cli/__init__.py

@@ -0,0 +1,11 @@
+# standard imports
+import os
+
+# local imports
+from .arg import process_flags
+from .config import process_config
+
+
+__script_dir = os.path.dirname(os.path.realpath(__file__))
+data_dir = os.path.join(os.path.dirname(__script_dir), 'data')
+config_dir = os.path.join(data_dir, 'config')

chainqueue/cli/arg.py

@@ -0,0 +1,2 @@
+def process_flags(argparser, flags):
+    argparser.add_argument('--backend', type=str, help='Backend to use for state store')

chainqueue/cli/config.py

@@ -0,0 +1,8 @@
+def process_config(config, args, flags):
+        args_override = {}
+
+        args_override['QUEUE_BACKEND'] = getattr(args, 'backend')
+
+        config.dict_override(args_override, 'local cli args')
+
+        return config

chainqueue/cli/output.py

@@ -0,0 +1,153 @@
+# standard imports
+import logging
+import enum
+
+# external imports
+from hexathon import add_0x
+
+# local imports
+from chainqueue.enum import (
+        StatusBits,
+        all_errors,
+        is_alive,
+        is_error_status,
+        status_str,
+        )
+
+logg = logging.getLogger(__name__)
+
+
+class OutputCol(enum.Enum):
+    chainspec = 0
+    hash = 1
+    statustext = 2
+    statuscode = 3
+    signedtx = 4
+
+
+class Outputter:
+    """Output helper for chainqueue cli listings tools.
+
+    :param chain_spec: Chain spec to use as getter context
+    :type chain_spec: chainlib.chain.ChainSpec
+    :param writer: Writer to write output to. Will automatically flush.
+    :type writer: Writer
+    :param getter: Transaction getter
+    :type getter: See chainqueue.sql.backend.get_otx 
+    :param session_method: Backend session generator method
+    :type session_method: varies
+    :param decode_status: Print status bit details
+    :type decode_status: bool
+    """
+
+    all_cols = [
+            OutputCol.chainspec,
+            OutputCol.hash,
+            OutputCol.signedtx,
+            OutputCol.statustext,
+            OutputCol.statuscode,
+            ]
+    default_cols = [
+            OutputCol.chainspec,
+            OutputCol.hash,
+            OutputCol.statustext,
+            OutputCol.statuscode,
+            ]
+
+    def __init__(self, chain_spec, writer, getter, session_method=None, decode_status=True, cols=None):
+        self.decode_status = decode_status
+        self.writer = writer
+        self.getter = getter
+        self.chain_spec = chain_spec
+        self.chain_spec_str = str(chain_spec)
+        self.session = None
+        if session_method != None:
+            self.session = session_method()
+        self.results = {
+            'pending_error': 0,
+            'final_error': 0,
+            'pending': 0,
+            'final': 0,
+                }
+
+        debug_col_name = []
+        if cols == None:
+            self.cols = Outputter.default_cols
+        else:
+            self.cols = []
+            for col in cols:
+                v = getattr(OutputCol, col)
+                self.cols.append(v) 
+
+        for col in self.cols:
+            debug_col_name.append(col.name)
+        logg.debug('outputter initialized with cols: {}'.format(','.join(debug_col_name)))
+
+
+    def __del__(self):
+        if self.session != None:
+            self.session.close()
+
+
+    def add(self, tx_hash):
+        """Retrieve a transaction by hash and add it for summary output generation.
+
+        :param tx_hash: Transaction hash
+        :type tx_hash: str
+        """
+        tx = self.getter(self.chain_spec, tx_hash, session=self.session)
+        self.__add(tx)
+
+    def __add(self, tx):
+        category = None
+        if is_alive(tx['status_code']):
+            category = 'pending'
+        else:
+            category = 'final'
+        self.results[category] += 1
+        if is_error_status(tx['status_code']):
+            logg.debug('registered {} as {} with error'.format(tx['tx_hash'], category))
+            self.results[category + '_error'] += 1
+        else:
+            logg.debug('registered {} as {}'.format(tx['tx_hash'], category))
+     
+
+    def decode_summary(self):
+        """Writes summary to the registered writer.
+        """
+        self.writer.write('pending\t{}\t{}\n'.format(self.results['pending'], self.results['pending_error']))
+        self.writer.write('final\t{}\t{}\n'.format(self.results['final'], self.results['final_error']))
+        self.writer.write('total\t{}\t{}\n'.format(self.results['final'] + self.results['pending'], self.results['final_error'] + self.results['pending_error']))
+
+
+    def decode_single(self, tx_hash):
+        """Retrieves the transaction with the given hash and writes the details to the underlying writer.
+
+        Registers the transaction with the summary generator.
+
+        :param tx_hash: Transaction hash
+        :type tx_hash: str
+        """
+        tx = self.getter(self.chain_spec, tx_hash, session=self.session)
+        self.__add(tx)
+        status = tx['status']
+        if self.decode_status:
+            status = status_str(tx['status_code'], bits_only=True)
+
+        vals = [
+            self.chain_spec_str,
+            add_0x(tx_hash),
+            status,
+            str(tx['status_code']),
+            add_0x(tx['signed_tx']),
+            ]
+
+        i = 0
+        l = len(self.cols)
+        for col in self.cols:
+            self.writer.write(vals[col.value])
+            i += 1
+            if i == l:
+                self.writer.write('\n')
+            else:
+                self.writer.write('\t')

chainqueue/data/config/config.ini

@@ -1,9 +1,2 @@
-[database]
-name =
-engine =
-driver =
-host =
-port =
-user =
-password =
-debug = 0
+[queue]
+backend = mem

chainqueue/entry.py

@@ -134,6 +134,10 @@ class QueueEntry:
             self.store.cache.set_block(self.tx_hash, block, tx)
 
 
+    def test(self, state):
+        return self.__match_state(state)
+
+
     def __str__(self):
         v = self.store.get(self.tx_hash)
         n = self.store.state(v[0])

chainqueue/error.py

@@ -24,12 +24,6 @@ class CacheIntegrityError(ChainQueueException):
     pass
 
 
-class BackendIntegrityError(ChainQueueException):
-    """Raised when queue backend has invalid state
-    """
-    pass
-
-
 class DuplicateTxError(ChainQueueException):
     """Backend already knows transaction
     """

chainqueue/runnable/list.py

@@ -12,7 +12,7 @@ from chainlib.chain import ChainSpec
 from crypto_dev_signer.eth.signer import ReferenceSigner as EIP155Signer
 
 # local imports
-from chainqueue.cli import Outputter
+from chainqueue.cli.output import Outputter
 
 
 logging.basicConfig(level=logging.WARNING)

chainqueue/settings.py

@@ -0,0 +1,8 @@
+# external imports
+from chainlib.settings import ChainSettings
+
+
+class ChainqueueSettings(ChainSettings):
+
+    def process_queue_backend(self, config):
+        self.o['QUEUE_BACKEND'] = config.get('QUEUE_BACKEND')

chainqueue/state.py

@@ -106,10 +106,10 @@ class Verify:
 
 class Status(shep.persist.PersistedState):
    
-    def __init__(self, store_factory):
+    def __init__(self, store_factory, allow_invalid=False, event_callback=None):
         verify = Verify().verify
         self.set_default_state('PENDING')
-        super(Status, self).__init__(store_factory, 12, verifier=verify)
+        super(Status, self).__init__(store_factory, 12, verifier=verify, check_alias=not allow_invalid, event_callback=event_callback)
         self.add('QUEUED')
         self.add('RESERVED')
         self.add('IN_NETWORK')

chainqueue/store/base.py

@@ -2,12 +2,15 @@
 import re
 import datetime
 import logging
+import time
 
 # local imports
 from chainqueue.cache import CacheTx
 from chainqueue.entry import QueueEntry
-from chainqueue.error import (
-        NotLocalTxError,
+from chainqueue.error import NotLocalTxError
+from chainqueue.enum import (
+        StatusBits,
+        all_errors,
         )
 
 logg = logging.getLogger(__name__)
@@ -21,11 +24,12 @@ def from_key(k):
     (ts_str, seq_str, tx_hash) = k.split('_')
     return (float(ts_str), int(seq_str), tx_hash, )
 
+all_local_errors = all_errors() - StatusBits.NETWORK_ERROR
 
 re_u = r'^[^_][_A-Z]+$'
 class Store:
 
-    def __init__(self, chain_spec, state_store, index_store, counter, cache=None):
+    def __init__(self, chain_spec, state_store, index_store, counter, cache=None, sync=True):
         self.chain_spec = chain_spec
         self.cache = cache
         self.state_store = state_store
@@ -43,9 +47,21 @@ class Store:
                 'unset',
                 'name',
                 'modified',
+                'purge',
                 ]:
             setattr(self, v, getattr(self.state_store, v))
-        self.state_store.sync()
+
+        if not sync:
+            return
+
+        sync_err = None
+        try:
+            self.state_store.sync()
+        except Exception as e:
+            sync_err = e
+
+        if sync_err != None:
+            raise FileNotFoundError(sync_err)
 
 
     def put(self, v, cache_adapter=CacheTx):
@@ -63,29 +79,41 @@ class Store:
 
 
     def get(self, k):
+        v = None
+        s = self.index_store.get(k)
+        err = None
         try:
-            s = self.index_store.get(k)
-        except FileNotFoundError:
-            raise NotLocalTxError(k)
-        v = self.state_store.get(s)
+            v = self.state_store.get(s)
+        except FileNotFoundError as e:
+            err = e
+        if v == None:
+            raise NotLocalTxError('could not find tx {}: {}'.format(k, err))
         return (s, v,)
 
 
-    def by_state(self, state=0, limit=4096, strict=False, threshold=None):
+    def by_state(self, state=0, not_state=0, limit=4096, strict=False, threshold=None):
         hashes = []
         i = 0
 
         refs_state = self.state_store.list(state)
+        refs_state.sort()
         
         for ref in refs_state:
             v = from_key(ref)
             hsh = v[2]
 
+            item_state = self.state_store.state(ref)
+
             if strict:
-                item_state = self.state_store.state(ref)
                 if item_state & state != item_state:
                     continue
 
+            if item_state & not_state > 0:
+                continue
+
+            item_state_str = self.state_store.name(item_state)
+            logg.info('state {} {} ({})'.format(ref, item_state_str, item_state))
+
             if threshold != None:
                 v = self.state_store.modified(ref)
                 if v > threshold:
@@ -93,7 +121,9 @@ class Store:
 
             hashes.append(hsh)
 
-
+            i += 1
+            if limit > 0 and i == limit:
+                break
 
         hashes.sort()
         return hashes
@@ -107,6 +137,17 @@ class Store:
         return self.by_state(state=self.DEFERRED, limit=limit, threshold=threshold)
 
 
+    def failed(self, limit=4096):
+        #return self.by_state(state=all_local_errors, limit=limit)
+        r = []
+        r += self.by_state(state=self.LOCAL_ERROR, limit=limit)
+        r += self.by_state(state=self.NODE_ERROR, limit=limit)
+        r.sort()
+        if len(r) > limit:
+            r = r[:limit]
+        return r
+
+
     def pending(self, limit=4096):
         return self.by_state(state=0, limit=limit, strict=True)
 
@@ -129,6 +170,7 @@ class Store:
     def fail(self, k):
         entry = QueueEntry(self, k)
         entry.load()
+        logg.debug('fail {}'.format(k))
         entry.sendfail()
 
 
@@ -152,3 +194,13 @@ class Store:
         entry = QueueEntry(self, k)
         entry.load()
         entry.sent()
+
+
+    def is_reserved(self, k):
+        entry = QueueEntry(self, k)
+        entry.load()
+        return entry.test(self.RESERVED)
+
+
+    def sync(self):
+        self.state_store.sync()

chainqueue/store/fs.py

@@ -6,7 +6,10 @@ import logging
 from leveldir.hex import HexDir
 
 # local imports
-from chainqueue.error import DuplicateTxError
+from chainqueue.error import (
+        DuplicateTxError,
+        NotLocalTxError,
+        )
 
 logg = logging.getLogger(__name__)
 
@@ -22,7 +25,7 @@ class IndexStore(HexDir):
         existing = None
         try:
             existing = self.get(k)
-        except FileNotFoundError:
+        except NotLocalTxError:
             pass
         return existing != None
 
@@ -37,7 +40,14 @@ class IndexStore(HexDir):
 
     def get(self, k):
         fp = self.store.to_filepath(k)
-        f = open(fp, 'rb')
+        f = None
+        err = None
+        try:
+            f = open(fp, 'rb')
+        except FileNotFoundError as e:
+            err = e
+        if err != None:
+            raise NotLocalTxError(err)
         v = f.read()
         f.close()
         return v.decode('utf-8')
@@ -64,7 +74,7 @@ class CounterStore:
     
         v = f.read(8)
         self.count = int.from_bytes(v, byteorder='big')
-        logg.info('counter starts at {}'.format(self.count))
+        logg.debug('counter starts at {}'.format(self.count))
     
         f.seek(0)
 

requirements.txt

@@ -1,9 +1,9 @@
-pysha3==1.0.2
-hexathon~=0.1.5
+#pysha3==1.0.2
+hexathon~=0.1.6
 leveldir~=0.3.0
-alembic==1.4.2
-SQLAlchemy==1.3.20
+#alembic==1.4.2
+#SQLAlchemy==1.3.20
 confini~=0.6.0
-pyxdg~=0.27
-chainlib>=0.1.0b1,<=0.1.0
-shep>=0.1.1rc1,<=0.2.0
+#pyxdg~=0.27
+chainlib~=0.1.1
+shep~=0.2.8

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = chainqueue
-version = 0.1.0
+version = 0.1.15
 description = Generic blockchain transaction queue control
 author = Louis Holbrook
 author_email = dev@holbrook.no
@@ -25,7 +25,7 @@ licence_files =
 	LICENSE.txt
 
 [options]
-python_requires = >= 3.6
+python_requires = >= 3.7
 include_package_data = True
 packages = 
 	chainqueue
@@ -33,6 +33,7 @@ packages =
 	chainqueue.unittest
 	chainqueue.store
 	chainqueue.runnable
+	chainqueue.cli
 
 #[options.entry_points]
 #console_scripts =

tests/test_store.py

@@ -6,14 +6,23 @@ import logging
 import shutil
 
 # external imports
+from chainlib.chain import ChainSpec
+from shep.store.noop import NoopStoreFactory
 
 # local imports
 from chainqueue.store.fs import (
         IndexStore,
         CounterStore,
         )
+from chainqueue.store.base import Store
 from chainqueue.error import DuplicateTxError
+from chainqueue.state import Status
 
+# tests imports
+from tests.common import (
+        MockTokenCache,
+        MockCacheTokenTx,
+        )
 
 logging.basicConfig(level=logging.DEBUG)
 logg = logging.getLogger()
@@ -58,5 +67,38 @@ class TestStoreImplementations(unittest.TestCase):
             store.put(hx, data)
 
 
+    def test_upcoming_limit(self):
+        index_store = IndexStore(self.path)
+        counter_store = CounterStore(self.path)
+        chain_spec = ChainSpec('foo', 'bar', 42, 'baz') 
+        factory = NoopStoreFactory().add
+        state_store = Status(factory)
+        cache_store = MockTokenCache()
+        queue_store = Store(chain_spec, state_store, index_store, counter_store, cache=cache_store)
+
+        txs = []
+        for i in range(3):
+            tx_src = os.urandom(128).hex()
+            tx = queue_store.put(tx_src, cache_adapter=MockCacheTokenTx)
+            txs.append(tx)
+
+        r = queue_store.upcoming(limit=3)
+        self.assertEqual(len(r), 0)
+
+        for tx in txs:
+            queue_store.enqueue(tx[1])
+
+        r = queue_store.upcoming(limit=3)
+        self.assertEqual(len(r), 3)
+
+        queue_store.send_start(txs[0][1])
+        r = queue_store.upcoming(limit=3)
+        self.assertEqual(len(r), 2)
+
+        queue_store.send_end(txs[0][1])
+        r = queue_store.upcoming(limit=3)
+        self.assertEqual(len(r), 2)
+
+
 if __name__ == '__main__':
     unittest.main()