Allow backend objects to move between sync and get

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

@@ -8,6 +8,11 @@ from chainqueue.cache import CacheTx
 from chainqueue.entry import QueueEntry
 from chainqueue.error import (
         NotLocalTxError,
+        BackendIntegrityError,
+        )
+from chainqueue.enum import (
+        StatusBits,
+        all_errors,
         )
 
 logg = logging.getLogger(__name__)
@@ -21,6 +26,7 @@ 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:
@@ -45,7 +51,17 @@ class Store:
                 'modified',
                 ]:
             setattr(self, v, getattr(self.state_store, v))
-        self.state_store.sync()
+
+        sync_err = None
+        for i in range(2):
+            try:
+                self.state_store.sync()
+            except Exception as e:
+                sync_err = e
+                continue
+
+        if sync_err != None:
+            raise BackendIntegrityError(sync_err)
 
 
     def put(self, v, cache_adapter=CacheTx):
@@ -63,30 +79,41 @@ class Store:
 
 
     def get(self, k):
-        try:
+        v = None
+        for i in range(2):
             s = self.index_store.get(k)
-        except FileNotFoundError:
+            try:
+                self.state_store.sync()
+                v = self.state_store.get(s)
+            except FileNotFoundError:
+                continue
+            break
+        if v == None:
             raise NotLocalTxError(k)
-        self.state_store.sync()
-        v = self.state_store.get(s)
         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
 
+            logg.info('state {} {}'.format(ref, item_state))
+            if item_state & not_state > 0:
+                continue
+
             if threshold != None:
                 v = self.state_store.modified(ref)
                 if v > threshold:
@@ -94,7 +121,9 @@ class Store:
 
             hashes.append(hsh)
 
-
+            i += 1
+            if limit > 0 and i == limit:
+                break
 
         hashes.sort()
         return hashes
@@ -108,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)
 
@@ -130,6 +170,7 @@ class Store:
     def fail(self, k):
         entry = QueueEntry(self, k)
         entry.load()
+        logg.debug('fail {}'.format(k))
         entry.sendfail()
 
 
@@ -159,3 +200,7 @@ class Store:
         entry = QueueEntry(self, k)
         entry.load()
         return entry.test(self.RESERVED)
+
+
+    def sync(self):
+        self.state_store.sync()

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = chainqueue
-version = 0.1.5
+version = 0.1.6
 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()