Add state finalizers

chainqueue/cache/__init__.py

@@ -0,0 +1 @@
+from .base import *

chainqueue/cache/base.py

@@ -29,7 +29,7 @@ class CacheTx:
         self.nonce = None
         self.value = None
 
-        self.tx_hash = None
+        self.hash = None
         self.block_number = None
         self.tx_index = None
         self.timestamp = None
@@ -42,7 +42,7 @@ class CacheTx:
 
 
     def init(self, tx_hash, nonce, sender, recipient, value):
-        self.tx_hash = self.normalizer.hash(tx_hash)
+        self.hash = self.normalizer.hash(tx_hash)
         self.sender = self.normalizer.address(sender)
         self.recipient = self.normalizer.address(recipient)
         self.nonce = nonce
@@ -59,7 +59,7 @@ class CacheTx:
 
 
     def __str__(self):
-        return '{}: {} ({}) -> {} = {}'.format(self.tx_hash, self.sender, self.nonce, self.recipient, self.value)
+        return '{}: {} ({}) -> {} = {}'.format(self.hash, self.sender, self.nonce, self.recipient, self.value)
 
 
 
@@ -133,3 +133,7 @@ class Cache:
 
     def count(self, cache_filter=None):
         raise NotImplementedError()
+
+
+    def set_block(self, block, tx):
+        raise NotImplementedError()

chainqueue/cache/fs.py

@@ -0,0 +1,10 @@
+# local imports
+from .base import Cache
+
+
+class FsCache(Cache):
+
+    def __init__(self, path):
+        self.path = path
+
+

chainqueue/entry.py

@@ -27,6 +27,10 @@ class QueueEntry:
         self.synced = False
 
 
+    def serialize(self):
+        return self.signed_tx
+
+
     def create(self, signed_tx):
         signed_tx = normalize_hex(signed_tx)
         self.k = self.store.put(self.tx_hash, signed_tx)
@@ -101,12 +105,12 @@ class QueueEntry:
         self.store.change(self.k, self.store.RESERVED, self.store.QUEUED) 
 
 
-    def fail(self, block):
+    def fail(self, block, tx):
         if self.__match_state(self.store.NETWORK_ERROR):
             return
         self.store.set(self.k, self.store.NETWORK_ERROR)
-        if self.cache:
-            self.cache.set_block(self.tx_hash, block)
+        if self.store.cache:
+            self.store.cache.set_block(self.tx_hash, block, tx)
 
 
     def cancel(self, confirmed=False):
@@ -116,5 +120,14 @@ class QueueEntry:
             self.store.change(self.k, self.store.OBSOLETE, self.store.RESERVED | self.store.QUEUED)
 
 
-    def succeed(self, block):
+    def succeed(self, block, tx):
         self.store.set(self.k, self.store.FINAL)
+        if self.store.cache:
+            self.store.cache.set_block(self.tx_hash, block, tx)
+
+
+    def __str__(self):
+        v = self.store.get(self.tx_hash)
+        n = self.store.state(v[0])
+        s = self.store.name(n)
+        return '{}: {}'.format(self.tx_hash, s)

chainqueue/state.py

@@ -46,7 +46,7 @@ class Verify:
             return 'already finalized'
         if from_state & state_store.IN_NETWORK:
             return 'already in network'
-        if from_state & state_store.RESERVED:
+        if not from_state & state_store.RESERVED:
             return 'not reserved'
         if from_state & state_store.mask_error:
             return 'already in error state'

chainqueue/store.py

@@ -1,81 +0,0 @@
-# standard imports
-import logging
-import re
-import datetime
-
-# local imports
-from chainqueue.cache import CacheTx
-
-
-logg = logging.getLogger(__name__)
-
-
-def to_key(t, n, k):
-    return '{}_{}_{}'.format(t, n, k)
-
-
-def from_key(k):
-    (ts_str, seq_str, tx_hash) = k.split('_')
-    return (float(ts_str), int(seq_str), tx_hash, )
-
-
-re_u = r'^[^_][_A-Z]+$'
-class Store:
-
-    def __init__(self, chain_spec, state_store, index_store, counter, cache=None):
-        self.chain_spec = chain_spec
-        self.cache = cache
-        self.state_store = state_store
-        self.index_store = index_store
-        self.counter = counter
-        for s in dir(self.state_store):
-            if not re.match(re_u, s):
-                continue
-            v = self.state_store.from_name(s)
-            setattr(self, s, v)
-        for v in ['state', 'change', 'set', 'unset']:
-            setattr(self, v, getattr(self.state_store, v))
-
-
-    def put(self, k, v, cache_adapter=CacheTx):
-        n = self.counter.next()
-        t = datetime.datetime.now().timestamp()
-        s = to_key(t, n, k)
-        self.state_store.put(s, v)
-        self.index_store.put(k, s)
-        if self.cache != None:
-            tx = cache_adapter()
-            tx.deserialize(v)
-            self.cache.put(self.chain_spec, tx) 
-
-
-    def get(self, k):
-        s = self.index_store.get(k)
-        v = self.state_store.get(s)
-        return (s, v,)
-
-
-    def by_state(self, state=0, limit=4096, strict=False):
-        hashes = []
-        i = 0
-
-        hashes_state = self.state_store.list(state)
-        if strict:
-            for k in hashes_state:
-                item_state = self.state_store.state(k)
-                if item_state & state != item_state:
-                    continue
-                hashes.append(k)
-        else:
-            hashes = hashes_state
-
-        hashes.sort()
-        hashes_out = []
-        for h in hashes:
-            pair = from_key(h)
-            hashes_out.append(pair[1])
-        return hashes_out
-
-
-    def upcoming(self, limit=4096):
-        return self.by_state(state=self.QUEUED, limit=limit)

chainqueue/store/__init__.py

@@ -0,0 +1,5 @@
+from .base import (
+        to_key,
+        from_key,
+        Store,
+        )

chainqueue/store/base.py

@@ -0,0 +1,143 @@
+# standard imports
+import re
+import datetime
+
+# local imports
+from chainqueue.cache import CacheTx
+from chainqueue.entry import QueueEntry
+
+
+def to_key(t, n, k):
+    return '{}_{}_{}'.format(t, n, k)
+
+
+def from_key(k):
+    (ts_str, seq_str, tx_hash) = k.split('_')
+    return (float(ts_str), int(seq_str), tx_hash, )
+
+
+re_u = r'^[^_][_A-Z]+$'
+class Store:
+
+    def __init__(self, chain_spec, state_store, index_store, counter, cache=None):
+        self.chain_spec = chain_spec
+        self.cache = cache
+        self.state_store = state_store
+        self.index_store = index_store
+        self.counter = counter
+        for s in dir(self.state_store):
+            if not re.match(re_u, s):
+                continue
+            v = self.state_store.from_name(s)
+            setattr(self, s, v)
+        for v in [
+                'state',
+                'change',
+                'set',
+                'unset',
+                'name',
+                'modified',
+                ]:
+            setattr(self, v, getattr(self.state_store, v))
+
+
+    def put(self, k, v, cache_adapter=CacheTx):
+        n = self.counter.next()
+        t = datetime.datetime.now().timestamp()
+        s = to_key(t, n, k)
+        self.state_store.put(s, v)
+        self.index_store.put(k, s)
+        if self.cache != None:
+            tx = cache_adapter()
+            tx.deserialize(v)
+            self.cache.put(self.chain_spec, tx) 
+        return s
+
+
+    def get(self, k):
+        s = self.index_store.get(k)
+        v = self.state_store.get(s)
+        return (s, v,)
+
+
+    def by_state(self, state=0, limit=4096, strict=False, threshold=None):
+        hashes = []
+        i = 0
+
+        refs_state = self.state_store.list(state)
+        
+        for ref in refs_state:
+            v = from_key(ref)
+            hsh = v[2]
+
+            if strict:
+                item_state = self.state_store.state(ref)
+                if item_state & state != item_state:
+                    continue
+
+            if threshold != None:
+                v = self.state_store.modified(ref)
+                if v > threshold:
+                    continue
+
+            hashes.append(hsh)
+
+
+
+        hashes.sort()
+        return hashes
+
+
+    def upcoming(self, limit=4096):
+        return self.by_state(state=self.QUEUED, limit=limit)
+
+
+    def deferred(self, limit=4096, threshold=None):
+        return self.by_state(state=self.DEFERRED, limit=limit, threshold=threshold)
+
+
+    def pending(self, limit=4096):
+        return self.by_state(state=0, limit=limit, strict=True)
+
+
+    def reserve(self, k):
+        entry = QueueEntry(self, k)
+        entry.load()
+        entry.reserve()
+
+
+    def enqueue(self, k):
+        entry = QueueEntry(self, k)
+        entry.load()
+        try:
+            entry.retry()
+        except StateTransitionInvalid:
+            entry.readysend()
+
+
+    def fail(self, k):
+        entry = QueueEntry(self, k)
+        entry.load()
+        entry.sendfail()
+
+
+    def final(self, k, block, tx, error=False):
+        entry = QueueEntry(self, k)
+        entry.load()
+        if error:
+            entry.fail(block, tx)
+        else:
+            entry.succeed(block, tx)
+
+
+    def send_start(self, k):
+        entry = QueueEntry(self, k)
+        entry.load()
+        entry.reserve()
+        return entry
+
+
+    def send_end(self, k):
+        entry = QueueEntry(self, k)
+        entry.load()
+        entry.sent()

chainqueue/store/fs.py

@@ -0,0 +1,70 @@
+# standard imports
+import os
+import logging
+
+# external imports
+from leveldir.hex import HexDir
+
+logg = logging.getLogger(__name__)
+
+
+class IndexStore(HexDir):
+
+    def __init__(self, root_path, digest_bytes=32):
+        os.path.join(root_path, 'contents')
+        self.store = HexDir(root_path, digest_bytes)
+
+
+    def put(self, k, v):
+        kb = bytes.fromhex(k)
+        vb = v.encode('utf-8')
+        self.store.add(kb, vb)
+
+
+    def get(self, k):
+        fp = self.store.to_filepath(k)
+        f = open(fp, 'rb')
+        v = f.read()
+        f.close()
+        return v.decode('utf-8')
+
+
+class CounterStore:
+
+    def __init__(self, root_path):
+        try:
+            os.stat(root_path)
+        except FileNotFoundError:
+            os.makedirs(root_path)
+
+        fp = os.path.join(root_path, '.counter')
+        f = None
+        try:
+            f = open(fp, 'rb+')
+        except FileNotFoundError:
+            logg.debug('counter not found, creating new in {}'.format(fp))
+            f = open(fp, 'wb+')
+            f.write(b'\x00' * 8)
+            f.close()
+            f = open(fp, 'rb+')
+    
+        v = f.read(8)
+        self.count = int.from_bytes(v, byteorder='big')
+        logg.info('counter starts at {}'.format(self.count))
+    
+        f.seek(0)
+
+        self.f = f
+
+
+    def __del__(self):
+        self.f.close()
+
+
+    def next(self):
+        c = self.count
+        self.count += 1
+        v = self.count.to_bytes(8, 'big')
+        self.f.write(v)
+        self.f.seek(0)
+        return c

requirements.txt

@@ -6,4 +6,4 @@ SQLAlchemy==1.3.20
 confini~=0.6.0
 pyxdg~=0.27
 chainlib>=0.1.0b1,<=0.1.0
-shep~=0.1.1
+shep>=0.1.1rc1,<=0.2.0

tests/base_shep.py

@@ -1,6 +1,7 @@
 # standard imports
 import tempfile
 import unittest
+import shutil
 
 # external imports
 from shep.store.file import SimpleFileStoreFactory
@@ -13,22 +14,10 @@ from chainqueue import (
         )
 
 # test imports
-from tests.common import MockCounter
-
-
-
-class MockContentStore:
-
-    def __init__(self):
-        self.store = {}
-
-
-    def put(self, k, v):
-        self.store[k] = v
-
-
-    def get(self, k):
-        return self.store.get(k)
+from tests.common import (
+        MockCounter,
+        MockContentStore,
+        )
 
 
 class TestShepBase(unittest.TestCase):
@@ -41,3 +30,7 @@ class TestShepBase(unittest.TestCase):
         counter = MockCounter()
         chain_spec = ChainSpec('foo', 'bar', 42, 'baz')
         self.store = Store(chain_spec, self.state, content_store, counter)
+
+
+    def tearDown(self):
+        shutil.rmtree(self.path)

tests/common.py

@@ -27,7 +27,7 @@ class MockTokenCache(Cache):
         self.last_filter = None
 
     def put(self, chain_spec, cache_tx):
-        self.db[cache_tx.tx_hash] = cache_tx
+        self.db[cache_tx.hash] = cache_tx
 
 
     def get(self, chain_spec, tx_hash):
@@ -79,7 +79,6 @@ class MockCacheTokenTx(CacheTokenTx):
         z = h.digest()
         tx_hash = z.hex()
 
-        #tx = CacheTokenTx(normalizer=self.normalizer)
         self.init(tx_hash, nonce, sender, recipient, value)
         self.set('src_token', token)
         self.set('dst_token', token)
@@ -90,4 +89,15 @@ class MockCacheTokenTx(CacheTokenTx):
         return self
 
 
+class MockContentStore:
 
+    def __init__(self):
+        self.store = {}
+
+
+    def put(self, k, v):
+        self.store[k] = v
+
+
+    def get(self, k):
+        return self.store.get(k)

tests/test_cache.py

@@ -57,7 +57,7 @@ class TestCache(TestShepBase):
         self.assertTrue(isinstance(tx.value, float))
         self.assertEqual(tx.sender[:4], 'addr')
         self.assertEqual(tx.recipient[:4], 'addr')
-        self.assertEqual(tx.tx_hash[:11], 'ashbashhash')
+        self.assertEqual(tx.hash[:11], 'ashbashhash')
 
 
     def test_cache_putget(self):
@@ -65,7 +65,7 @@ class TestCache(TestShepBase):
         tx = MockCacheTokenTx()
         tx.deserialize(a)
         self.cache.put(self.chain_spec, tx)
-        tx_retrieved = self.cache.get(self.chain_spec, tx.tx_hash)
+        tx_retrieved = self.cache.get(self.chain_spec, tx.hash)
         self.assertEqual(tx, tx_retrieved)
 
 

tests/test_entry.py

@@ -44,7 +44,7 @@ class TestEntry(TestShepBase):
         txs = self.store.by_state(state=self.store.IN_NETWORK)
         self.assertEqual(len(txs), 1)
 
-        entry.succeed(0)
+        entry.succeed(None, None)
         txs = self.store.by_state()
         self.assertEqual(len(txs), 1)
       

tests/test_integrate.py

@@ -1,7 +1,9 @@
 # standard imports
+import os
 import tempfile
 import unittest
 import logging
+import time
 
 # external imports
 from shep.store.file import SimpleFileStoreFactory
@@ -18,6 +20,7 @@ from tests.common import (
         MockCounter,
         MockTokenCache,
         MockCacheTokenTx,
+        MockContentStore,
         )
 from tests.base_shep import TestShepBase
 
@@ -25,20 +28,6 @@ logging.basicConfig(level=logging.DEBUG)
 logg = logging.getLogger()
 
 
-class MockContentStore:
-
-    def __init__(self):
-        self.store = {}
-
-
-    def put(self, k, v):
-        self.store[k] = v
-
-
-    def get(self, k):
-        return self.store.get(k)
-
-
 class TestIntegrateBase(TestShepBase):
 
     def setUp(self):
@@ -53,7 +42,84 @@ class TestIntegrateBase(TestShepBase):
 
 
     def test_integration_valid(self):
-        self.store.put(b'foo'.hex(), b'bar'.hex(), cache_adapter=MockCacheTokenTx)
+        self.store.put(os.urandom(4).hex(), os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+
+    
+    def test_state_default(self):
+        hx = os.urandom(4).hex()
+        self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        v = self.store.pending()
+        self.assertEqual(len(v), 1)
+        self.assertEqual(v[0], hx)
+
+
+    def test_state_enqueue(self):
+        hx = os.urandom(4).hex()
+        self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.get(hx)
+        self.store.enqueue(hx)
+        v = self.store.upcoming()
+        self.assertEqual(len(v), 1)
+        v = self.store.pending()
+        self.assertEqual(len(v), 0)
+
+
+    def test_state_defer(self):
+        hx = os.urandom(4).hex()
+        self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
+        self.store.fail(hx)
+        v = self.store.deferred()
+        self.assertEqual(len(v), 1)
+        self.assertEqual(v[0], hx)
+
+
+    def test_state_multiple(self):
+        hx = os.urandom(4).hex()
+        self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
+        self.store.fail(hx)
+        hx = os.urandom(8).hex()
+        self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
+        self.store.fail(hx)
+        v = self.store.deferred()
+        self.assertEqual(len(v), 2)
+
+
+    def test_state_multiple_sort(self):
+        hx = os.urandom(4).hex()
+        self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
+        self.store.fail(hx)
+        hx = os.urandom(4).hex()
+        self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.enqueue(hx)
+        hx = os.urandom(4).hex()
+        self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
+        self.store.fail(hx)
+        hx = os.urandom(4).hex()
+        self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        v = self.store.deferred()
+        self.assertEqual(len(v), 2)
+
+
+    def test_state_date_threshold(self):
+        hx = os.urandom(4).hex()
+        s = self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
+        self.store.fail(hx)
+        then = self.store.modified(s)
+        time.sleep(0.1)
+
+        hx = os.urandom(4).hex()
+        s = self.store.put(hx, os.urandom(8).hex(), cache_adapter=MockCacheTokenTx)
+        self.store.reserve(hx)
+        self.store.fail(hx)
+
+        v = self.store.deferred(threshold=then)
+        self.assertEqual(len(v), 1)
 
 
 if __name__ == '__main__':

tests/test_store.py

@@ -0,0 +1,52 @@
+# standard imports
+import os
+import tempfile
+import unittest
+import logging
+import shutil
+
+# external imports
+
+# local imports
+from chainqueue.store.fs import (
+        IndexStore,
+        CounterStore,
+        )
+
+
+logging.basicConfig(level=logging.DEBUG)
+logg = logging.getLogger()
+
+class TestStoreImplementations(unittest.TestCase):
+
+    def setUp(self):
+        self.path = tempfile.mkdtemp()
+      
+
+    def tearDown(self):
+        shutil.rmtree(self.path)
+
+    
+    def test_basic_index(self):
+        store = IndexStore(self.path)
+        hx = os.urandom(32).hex()
+        data = 'foo_bar_baz'
+        store.put(hx, data)
+        r = store.get(hx)
+        self.assertEqual(data, r)
+
+
+    def test_basic_counter(self):
+        store = CounterStore(self.path)
+        v = store.next()
+        self.assertEqual(v, 0)
+        v = store.next()
+        self.assertEqual(v, 1)
+
+        store = CounterStore(self.path)
+        v = store.next()
+        self.assertEqual(v, 2)
+
+
+if __name__ == '__main__':
+    unittest.main()