Raise correct error in index store exists check

CHANGELOG

@@ -1,3 +1,24 @@
+- 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

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/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 (
+from chainqueue.error import NotLocalTxError
-        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,30 +79,42 @@ class Store:
 
 
     def get(self, k):
+        v = None
+        s = self.index_store.get(k)
+        err = None
         try:
-            s = self.index_store.get(k)
+            self.state_store.sync()
-        except FileNotFoundError:
+            v = self.state_store.get(s)
-            raise NotLocalTxError(k)
+        except FileNotFoundError as e:
-        self.state_store.sync()
+            err = e
-        v = self.state_store.get(s)
+        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:
@@ -94,7 +122,9 @@ class Store:
 
             hashes.append(hsh)
 
-
+            i += 1
+            if limit > 0 and i == limit:
+                break
 
         hashes.sort()
         return hashes
@@ -108,6 +138,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)
 
@@ -130,6 +171,7 @@ class Store:
     def fail(self, k):
         entry = QueueEntry(self, k)
         entry.load()
+        logg.debug('fail {}'.format(k))
         entry.sendfail()
 
 
@@ -153,3 +195,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
+#pysha3==1.0.2
 hexathon~=0.1.5
 leveldir~=0.3.0
-alembic==1.4.2
+#alembic==1.4.2
-SQLAlchemy==1.3.20
+#SQLAlchemy==1.3.20
 confini~=0.6.0
-pyxdg~=0.27
+#pyxdg~=0.27
-chainlib>=0.1.0b1,<=0.1.0
+chainlib~=0.1.1
-shep>=0.1.1rc1,<=0.3.0
+shep~=0.2.5

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = chainqueue
-version = 0.1.3
+version = 0.1.12
 description = Generic blockchain transaction queue control
 author = Louis Holbrook
 author_email = dev@holbrook.no

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()