Remove spam logline

CHANGELOG

@@ -1,3 +1,12 @@
+- 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/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,6 +24,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 +49,15 @@ class Store:
                 'modified',
                 ]:
             setattr(self, v, getattr(self.state_store, v))
-        self.state_store.sync()
+
+        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 +75,40 @@ 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
 
+            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 +116,9 @@ class Store:
 
             hashes.append(hsh)
 
-
+            i += 1
+            if limit > 0 and i == limit:
+                break
 
         hashes.sort()
         return hashes
@@ -108,6 +132,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 +165,7 @@ class Store:
     def fail(self, k):
         entry = QueueEntry(self, k)
         entry.load()
+        logg.debug('fail {}'.format(k))
         entry.sendfail()
 
 
@@ -159,3 +195,7 @@ class Store:
         entry = QueueEntry(self, k)
         entry.load()
         return entry.test(self.RESERVED)
+
+
+    def sync(self):
+        self.state_store.sync()

chainqueue/store/fs.py

@@ -64,7 +64,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)
 

setup.cfg

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