Upgrade shep

CHANGELOG

@@ -1,3 +1,12 @@
+* 0.3.5
+	- Allow memory-only shep if factory set to None in store constructor
+* 0.3.4
+	- Use explicit bool check in filter interrupt check
+* 0.3.3
+	- Include shep persistent state bootstrap sync
+	- Add chainsyncer extras
+* 0.3.2
+	- Implement rocksdb backend
 * 0.3.1
 	- Upgrade to release shep version
 	- Move sync state to SYNC after start

chainsyncer/session.py

@@ -1,9 +1,12 @@
 # standard imports
 import uuid
+import logging
 
 # local imports
 from chainsyncer.error import FilterDone
 
+logg = logging.getLogger(__name__)
+
 
 class SyncSession:
 
@@ -29,6 +32,7 @@ class SyncSession:
     def filter(self, conn, block, tx):
         self.session_store.connect()
         for fltr in self.filters:
+            logg.debug('executing filter {}'.format(fltr))
             self.item.advance()
             interrupt = fltr.filter(conn, block, tx)
             if not self.item.release(interrupt=interrupt):

chainsyncer/store/base.py

@@ -1,8 +1,10 @@
 # standard imports
+import os
 import logging
 
 # local imports
 from shep.persist import PersistedState
+from shep import State
 from shep.error import StateInvalid
 from chainsyncer.filter import FilterState
 from chainsyncer.error import (
@@ -120,7 +122,7 @@ class SyncItem:
     def release(self, interrupt=False):
         if self.skip_filter:
             return False
-        if interrupt:
+        if interrupt == True:
             self.filter_state.unset(self.state_key, self.filter_state.from_name('LOCK'))
             self.filter_state.set(self.state_key, self.filter_state.from_name('INTERRUPT'))
             self.filter_state.set(self.state_key, self.filter_state.from_name('DONE'))
@@ -145,7 +147,7 @@ class SyncItem:
 
 class SyncStore:
 
-    def __init__(self, session_id=None):
+    def __init__(self, path, session_id=None):
         self.session_id = None
         self.session_path = None
         self.is_default = False
@@ -155,17 +157,34 @@ class SyncStore:
         self.item_keys = []
         self.started = False
         self.thresholds = []
+        self.default_path = os.path.join(path, 'default')
+
+        if session_id == None:
+            self.session_path = os.path.realpath(self.default_path)
+            self.is_default = True
+        else:
+            if session_id == 'default':
+                self.is_default = True
+            given_path = os.path.join(path, session_id)
+            self.session_path = os.path.realpath(given_path)
 
 
-    def setup_sync_state(self, factory, event_callback):
-        self.state = PersistedState(factory.add, 2, event_callback=event_callback)
+    def setup_sync_state(self, factory=None, event_callback=None):
+        if factory == None:
+            self.state = State(2, event_callback=event_callback)
+        else:
+            self.state = PersistedState(factory.add, 2, event_callback=event_callback)
         self.state.add('SYNC')
         self.state.add('DONE')
 
 
-    def setup_filter_state(self, factory, event_callback):
-        filter_state_backend = PersistedState(factory.add, 0, check_alias=False, event_callback=event_callback)
-        self.filter_state = FilterState(filter_state_backend, scan=factory.ls)
+    def setup_filter_state(self, factory=None, event_callback=None):
+        if factory == None:
+            filter_state_backend = State(0, check_alias=False, event_callback=event_callback)
+            self.filter_state = FilterState(filter_state_backend)
+        else:
+            filter_state_backend = PersistedState(factory.add, 0, check_alias=False, event_callback=event_callback)
+            self.filter_state = FilterState(filter_state_backend, scan=factory.ls)
         self.filters = []
 
 
@@ -191,7 +210,7 @@ class SyncStore:
         if self.first:
             state_bytes = sync_state_serialize(offset, 0, target)
             block_number_str = str(offset)
-            self.state.put(block_number_str, state_bytes)
+            self.state.put(block_number_str, contents=state_bytes)
             self.filter_state.put(block_number_str)
             o = SyncItem(offset, target, self.state, self.filter_state)
             self.items[offset] = o
@@ -215,9 +234,7 @@ class SyncStore:
             self.state.move(item.state_key, self.state.DONE)
 
             state_bytes = sync_state_serialize(item.cursor, 0, -1)
-            self.state.put(str(item.cursor), state_bytes)
-
-        logg.debug('item {}'.format(self.state.state(item.state_key)))
+            self.state.put(str(item.cursor), contents=state_bytes)
 
 
     def load(self, target):

chainsyncer/store/fs.py

@@ -7,10 +7,7 @@ import logging
 from shep.store.file import SimpleFileStoreFactory
 
 # local imports 
-from chainsyncer.store import (
-        SyncItem,
-        SyncStore,
-        )
+from chainsyncer.store import SyncStore
 
 logg = logging.getLogger(__name__)
 
@@ -18,18 +15,7 @@ logg = logging.getLogger(__name__)
 class SyncFsStore(SyncStore):
 
     def __init__(self, base_path, session_id=None, state_event_callback=None, filter_state_event_callback=None):
-        super(SyncFsStore, self).__init__(session_id=session_id)
-
-        default_path = os.path.join(base_path, 'default')
-
-        if session_id == None:
-            self.session_path = os.path.realpath(default_path)
-            self.is_default = True
-        else:
-            if session_id == 'default':
-                self.is_default = True
-            given_path = os.path.join(base_path, session_id)
-            self.session_path = os.path.realpath(given_path)
+        super(SyncFsStore, self).__init__(base_path, session_id=session_id)
 
         create_path = False
         try:
@@ -38,9 +24,9 @@ class SyncFsStore(SyncStore):
             create_path = True
 
         if create_path:
-            self.__create_path(base_path, default_path, session_id=session_id)
-        self.session_id = os.path.basename(self.session_path)
+            self.__create_path(base_path, self.default_path, session_id=session_id)
 
+        self.session_id = os.path.basename(self.session_path)
         logg.info('session id {} resolved {} path {}'.format(session_id, self.session_id, self.session_path))
 
         base_sync_path = os.path.join(self.session_path, 'sync')

chainsyncer/store/mem.py

@@ -0,0 +1,34 @@
+# standard imports
+import logging
+import os
+
+# external imports
+from shep import State
+
+# local imports 
+from chainsyncer.store import SyncStore
+
+logg = logging.getLogger(__name__)
+
+
+class SyncMemStore(SyncStore):
+
+    def __init__(self, session_id=None, state_event_callback=None, filter_state_event_callback=None):
+        super(SyncMemStore, self).__init__('/dev/null', session_id=session_id)
+
+        self.session_id = os.path.basename(self.session_path)
+        logg.info('session id {} resolved {} path {}'.format(session_id, self.session_id, self.session_path))
+
+        factory = None
+        self.setup_sync_state(factory, state_event_callback)
+
+        factory = None
+        self.setup_filter_state(factory, filter_state_event_callback)
+
+
+    def set_target(self, v):
+        self.target = int(v)
+
+
+    def get_target(self):
+        return self.target

chainsyncer/store/rocksdb.py

@@ -0,0 +1,66 @@
+# standard imports
+import uuid
+import os
+import logging
+
+# external imports
+from shep.store.rocksdb import RocksDbStoreFactory
+
+# local imports 
+from chainsyncer.store import (
+        SyncItem,
+        SyncStore,
+        )
+
+logg = logging.getLogger(__name__)
+
+
+class RocksDbStoreAdder:
+
+    def __init__(self, factory, prefix):
+        self.factory = factory
+        self.prefix = prefix
+
+
+    def add(self, k):
+        path = os.path.join(self.prefix, k)
+        return self.factory.add(path)
+
+
+    def ls(self):
+        return self.factory.ls()
+
+
+class SyncRocksDbStore(SyncStore):
+
+    def __init__(self, base_path, session_id=None, state_event_callback=None, filter_state_event_callback=None):
+        super(SyncRocksDbStore, self).__init__(base_path, session_id=session_id)
+
+        self.factory = RocksDbStoreFactory(self.session_path, binary=True)
+        prefix_factory = RocksDbStoreAdder(self.factory, 'sync')
+        self.setup_sync_state(prefix_factory, state_event_callback)
+
+        prefix_factory = RocksDbStoreAdder(self.factory, 'filter')
+        self.setup_filter_state(prefix_factory, filter_state_event_callback)
+
+        self.session_id = os.path.basename(self.session_path)
+        logg.info('session id {} resolved {} path {}'.format(session_id, self.session_id, self.session_path))
+
+        self.target_db = RocksDbStoreAdder(self.factory, '.stat').add('target')
+
+
+    def get_target(self):
+        v = self.target_db.get('target')
+        if v != None:
+            self.target = int(v)
+
+
+    def set_target(self, v):
+        self.target_db.put('target', str(v))
+        self.target = v
+
+
+    def stop(self, item):
+        if item != None:
+            super(SyncRocksDbStore, self).stop(item)
+        self.factory.close()

chainsyncer/unittest/base.py

@@ -267,3 +267,12 @@ class MockChainInterfaceConn(MockConn):
 
     def handle_receipt(self, hsh):
         return {}
+
+
+class MockItem:
+
+    def __init__(self, target, offset, cursor, state_key):
+        self.target = target
+        self.offset = offset
+        self.cursor = cursor
+        self.state_key = state_key

chainsyncer/unittest/db.py

@@ -1,64 +0,0 @@
-# standard imports
-import logging
-import os
-
-# external imports
-import alembic
-import alembic.config
-
-# local imports
-from chainsyncer.db.models.base import SessionBase
-from chainsyncer.db import dsn_from_config
-from chainsyncer.db.models.base import SessionBase
-
-logg = logging.getLogger(__name__)
-
-
-class ChainSyncerDb:
-    """SQLITE database setup for unit tests
-
-    :param debug: Activate sql level debug (outputs sql statements)
-    :type debug: bool
-    """
-
-    base = SessionBase
-
-    def __init__(self, debug=False):
-        config = {
-            'DATABASE_ENGINE': 'sqlite',
-            'DATABASE_DRIVER': 'pysqlite',
-            'DATABASE_NAME': 'chainsyncer.sqlite',
-                }
-        logg.debug('config {}'.format(config))
-
-        self.dsn = dsn_from_config(config)
-
-        self.base.poolable = False
-        self.base.transactional = False
-        self.base.procedural = False
-        self.base.connect(self.dsn, debug=debug) # TODO: evaluates to "true" even if string is 0
-
-        rootdir = os.path.join(os.path.dirname(os.path.dirname(__file__)), '..')
-        dbdir = os.path.join(rootdir, 'chainsyncer', 'db')
-        #migrationsdir = os.path.join(dbdir, 'migrations', config.get('DATABASE_ENGINE'))
-        migrationsdir = os.path.join(dbdir, 'migrations', 'default')
-        logg.info('using migrations directory {}'.format(migrationsdir))
-
-        ac = alembic.config.Config(os.path.join(migrationsdir, 'alembic.ini'))
-        ac.set_main_option('sqlalchemy.url', self.dsn)
-        ac.set_main_option('script_location', migrationsdir)
-
-        alembic.command.downgrade(ac, 'base')
-        alembic.command.upgrade(ac, 'head')
-
-
-    def bind_session(self, session=None):
-        """Create session using underlying session base
-        """
-        return self.base.bind_session(session)
-
-    
-    def release_session(self, session=None):
-        """Release session using underlying session base
-        """
-        return self.base.release_session(session)

chainsyncer/unittest/store.py

@@ -1,13 +1,12 @@
 # standard imports
-import unittest
-import tempfile
-import shutil
-import logging
-import stat
 import os
+import stat
+import unittest
+import shutil
+import tempfile
+import logging
 
 # local imports
-from chainsyncer.store.fs import SyncFsStore
 from chainsyncer.session import SyncSession
 from chainsyncer.error import (
         LockError,
@@ -15,74 +14,136 @@ from chainsyncer.error import (
         IncompleteFilterError,
         SyncDone,
         )
-from chainsyncer.unittest import MockFilter
+from chainsyncer.unittest import (
+        MockFilter,
+        MockItem,
+        )
 
-logging.basicConfig(level=logging.DEBUG)
-logg = logging.getLogger()
+logging.STATETRACE = 5
+logg = logging.getLogger(__name__)
+logg.setLevel(logging.STATETRACE)
 
 
-class TestFs(unittest.TestCase):
+def state_change_callback(k, old_state, new_state):
+    logg.log(logging.STATETRACE, 'state change: {} {} -> {}'.format(k, old_state, new_state)) 
+
+
+def filter_change_callback(k, old_state, new_state):
+    logg.log(logging.STATETRACE, 'filter change: {} {} -> {}'.format(k, old_state, new_state)) 
+
+
+class TestStoreBase(unittest.TestCase):
 
     def setUp(self):
         self.path = tempfile.mkdtemp()
+        self.store_factory = None
+        self.persist = True
+
+
+    @classmethod
+    def link(cls, target):
+        for v in [
+                "default",
+                "store_start",
+                "store_resume",
+                "sync_process_nofilter",
+                "sync_process_onefilter",
+                "sync_process_outoforder",
+                "sync_process_interrupt",
+                "sync_process_reset",
+                "sync_process_done",
+                "sync_head_future",
+                "sync_history_interrupted",
+                "sync_history_complete",
+                ]:
+            setattr(target, 'test_' + v, getattr(cls, 't_' + v))
 
 
     def tearDown(self):
         shutil.rmtree(self.path)
 
 
-    def test_default(self):
-        store = SyncFsStore(self.path)
+    def t_default(self):
+        bogus_item = MockItem(0, 0, 0, 0)
+        store = self.store_factory()
+        
         fp = os.path.join(self.path, store.session_id)
         session_id = store.session_id
-        st = os.stat(fp)
-        self.assertTrue(stat.S_ISDIR(st.st_mode))
-        self.assertTrue(store.is_default)
-        
+        st = None
+        try:
+            st = os.stat(fp)
+        except FileNotFoundError as e:
+            logg.warning('error {} persist {}'.format(e, self.persist))
+            if self.persist:
+                raise e
+
+        if st != None: 
+            self.assertTrue(stat.S_ISDIR(st.st_mode))
+            self.assertTrue(store.is_default)
+
         fpd = os.path.join(self.path, 'default')
-        st = os.stat(fpd)
-        self.assertTrue(stat.S_ISDIR(st.st_mode))
-        self.assertTrue(store.is_default)
+        try:
+            st = os.stat(fpd)
+        except FileNotFoundError as e:
+            logg.warning('error {} persist {}'.format(e, self.persist))
+            if self.persist:
+                raise e
+        if st != None:
+            self.assertTrue(stat.S_ISDIR(st.st_mode))
+            self.assertTrue(store.is_default)
 
         fpd = os.path.realpath(fpd)
         self.assertEqual(fpd, fp)
 
-        store = SyncFsStore(self.path)
+        store.stop(bogus_item)
+        store = self.store_factory()
         fpr = os.path.join(self.path, session_id)
         self.assertEqual(fp, fpr)
         self.assertTrue(store.is_default)
 
-        store = SyncFsStore(self.path, 'default')
+        store.stop(bogus_item)
+        store = self.store_factory('default')
         fpr = os.path.join(self.path, session_id)
         self.assertEqual(fp, fpr)
         self.assertTrue(store.is_default)
 
-        store = SyncFsStore(self.path, 'foo')
+        store.stop(bogus_item)
+        store = self.store_factory('foo')
         fpf = os.path.join(self.path, 'foo')
-        st = os.stat(fpf)
-        self.assertTrue(stat.S_ISDIR(st.st_mode))
-        self.assertFalse(store.is_default)
+        try:
+            st = os.stat(fpf)
+        except FileNotFoundError as e:
+            logg.warning('error {} persist {}'.format(e, self.persist))
+            if self.persist:
+                raise e
+        if st != None:
+            self.assertTrue(stat.S_ISDIR(st.st_mode))
+            self.assertFalse(store.is_default)
 
 
-    def test_store_start(self):
-        store = SyncFsStore(self.path)
+    def t_store_start(self):
+        bogus_item = MockItem(0, 0, 0, 0)
+        store = self.store_factory()
         store.start(42)
         self.assertTrue(store.first)
 
-        store = SyncFsStore(self.path)
-        store.start()
-        self.assertFalse(store.first)
+        store.stop(bogus_item)
+
+        if self.persist:
+            store = self.store_factory()
+            store.start()
+            self.assertFalse(store.first)
 
 
-    def test_store_resume(self):
-        store = SyncFsStore(self.path)
+    def t_store_resume(self):
+        store = self.store_factory()
         store.start(13)
         self.assertTrue(store.first)
         # todo not done
 
 
-    def test_sync_process_nofilter(self):
-        store = SyncFsStore(self.path)
+    def t_sync_process_nofilter(self):
+        store = self.store_factory()
         session = SyncSession(store)
         session.start()
         o = session.get(0)
@@ -90,8 +151,8 @@ class TestFs(unittest.TestCase):
             o.advance()
 
 
-    def test_sync_process_onefilter(self):
-        store = SyncFsStore(self.path)
+    def t_sync_process_onefilter(self):
+        store = self.store_factory()
         session = SyncSession(store)
 
         fltr_one = MockFilter('foo')
@@ -103,8 +164,8 @@ class TestFs(unittest.TestCase):
         o.release()
 
 
-    def test_sync_process_outoforder(self):
-        store = SyncFsStore(self.path)
+    def t_sync_process_outoforder(self):
+        store = self.store_factory()
         session = SyncSession(store)
 
         fltr_one = MockFilter('foo')
@@ -126,8 +187,8 @@ class TestFs(unittest.TestCase):
         o.release()
 
 
-    def test_sync_process_interrupt(self):
-        store = SyncFsStore(self.path)
+    def t_sync_process_interrupt(self):
+        store = self.store_factory()
         session = SyncSession(store)
 
         fltr_one = MockFilter('foo')
@@ -143,8 +204,8 @@ class TestFs(unittest.TestCase):
             o.advance()
 
 
-    def test_sync_process_reset(self):
-        store = SyncFsStore(self.path)
+    def t_sync_process_reset(self):
+        store = self.store_factory()
         session = SyncSession(store)
 
         fltr_one = MockFilter('foo')
@@ -170,8 +231,8 @@ class TestFs(unittest.TestCase):
         o.reset()
 
     
-    def test_sync_process_done(self):
-        store = SyncFsStore(self.path)
+    def t_sync_process_done(self):
+        store = self.store_factory()
         session = SyncSession(store)
 
         fltr_one = MockFilter('foo')
@@ -188,8 +249,8 @@ class TestFs(unittest.TestCase):
             o.next(advance_block=True)
 
 
-    def test_sync_head_future(self):
-        store = SyncFsStore(self.path, session_id='foo')
+    def t_sync_head_future(self):
+        store = self.store_factory('foo')
         session = SyncSession(store)
 
         session.start()
@@ -198,13 +259,18 @@ class TestFs(unittest.TestCase):
         o.next(advance_block=True)
         session.stop(o)
 
-        store = SyncFsStore(self.path, session_id='foo')
-        store.start()
-        o = store.get(2)
+        if self.persist:
+            store = self.store_factory('foo')
+            store.start()
+            o = store.get(2)
 
 
-    def test_sync_history_interrupted(self):
-        store = SyncFsStore(self.path, session_id='foo')
+    def t_sync_history_interrupted(self):
+        if not self.persist:
+            return
+
+        bogus_item = MockItem(0, 0, 0, 0)
+        store = self.store_factory('foo')
         session = SyncSession(store)
 
         session.start(target=13)
@@ -213,7 +279,8 @@ class TestFs(unittest.TestCase):
         o.next(advance_block=True)
         session.stop(o)
 
-        store = SyncFsStore(self.path, session_id='foo')
+        store.stop(bogus_item)
+        store = self.store_factory('foo')
         store.start()
         o = store.get(0)
         self.assertEqual(o.cursor, 2)
@@ -221,15 +288,15 @@ class TestFs(unittest.TestCase):
         o.next(advance_block=True)
         o.next(advance_block=True)
 
-        session.stop(o)
-        store = SyncFsStore(self.path, session_id='foo')
+        store.stop(bogus_item)
+        store = self.store_factory('foo')
         store.start()
         self.assertEqual(o.cursor, 4)
         self.assertEqual(o.target, 13) 
 
 
-    def test_sync_history_complete(self):
-        store = SyncFsStore(self.path, session_id='foo')
+    def t_sync_history_complete(self):
+        store = self.store_factory('foo')
         session = SyncSession(store)
 
         session.start(target=3)
@@ -239,7 +306,3 @@ class TestFs(unittest.TestCase):
         o.next(advance_block=True)
         with self.assertRaises(SyncDone):
             o.next(advance_block=True)
-
-
-if __name__ == '__main__':
-    unittest.main()

requirements.txt

@@ -1,5 +1,5 @@
 confini~=0.6.0
 semver==2.13.0
 hexathon~=0.1.5
-chainlib>=0.1.0b1,<=0.1.0
-shep>=0.2.0rc1,<0.3.0
+chainlib>=0.1.0b1,<0.2.0
+shep~=0.2.3

run_tests.sh

@@ -8,5 +8,12 @@ for f in `ls tests/*.py`; do
 		exit
 	fi
 done
+
+for f in `ls tests/store/*.py`; do
+	python $f
+	if [ $? -gt 0 ]; then
+		exit
+	fi
+done
 set +x
 set +e

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = chainsyncer
-version = 0.3.1
+version = 0.3.5
 description = Generic blockchain syncer driver
 author = Louis Holbrook
 author_email = dev@holbrook.no
@@ -28,7 +28,6 @@ packages =
 	chainsyncer.driver
 	chainsyncer.unittest
 	chainsyncer.store
-	chainsyncer.state
 
 #[options.package_data]
 #* =

setup.py

@@ -26,5 +26,7 @@ setup(
     install_requires=requirements,
     extras_require={
         'sql': sql_requirements,
+        'rocksdb': ['shep[rocksdb]~=0.2.2'],
+        'redis': ['shep[redis]~=0.2.2'],
     }
     )

tests/store/test_0_mem.py

@@ -0,0 +1,33 @@
+# standard imports
+import unittest
+import logging
+
+# external imports
+from shep import State
+
+# local imports
+from chainsyncer.store.mem import SyncMemStore
+from chainsyncer.unittest.store import TestStoreBase
+
+logging.basicConfig(level=logging.DEBUG)
+logg = logging.getLogger()
+
+
+class StoreFactory:
+
+    def create(self, session_id=None):
+        return SyncMemStore(session_id=session_id)
+
+
+class TestMem(TestStoreBase):
+
+    def setUp(self):
+        super(TestMem, self).setUp()
+        self.store_factory = StoreFactory().create
+        self.persist = False
+
+
+if __name__ == '__main__':
+    TestStoreBase.link(TestMem)
+    # Remove tests that test persistence of state
+    unittest.main()

tests/store/test_1_fs.py

@@ -0,0 +1,32 @@
+# standard imports
+import unittest
+import logging
+
+# local imports
+from chainsyncer.store.fs import SyncFsStore
+from chainsyncer.unittest.store import TestStoreBase
+
+logging.basicConfig(level=logging.DEBUG)
+logg = logging.getLogger()
+
+
+class StoreFactory:
+
+    def __init__(self, path):
+        self.path = path
+
+
+    def create(self, session_id=None):
+        return SyncFsStore(self.path, session_id=session_id)
+
+
+class TestFs(TestStoreBase):
+
+    def setUp(self):
+        super(TestFs, self).setUp()
+        self.store_factory = StoreFactory(self.path).create
+
+
+if __name__ == '__main__':
+    TestStoreBase.link(TestFs)
+    unittest.main()

tests/store/test_rocksdb.py

@@ -0,0 +1,35 @@
+# standard imports
+import unittest
+import logging
+
+# local imports
+from chainsyncer.store.rocksdb import SyncRocksDbStore
+from chainsyncer.unittest.store import (
+        TestStoreBase,
+        filter_change_callback,
+        state_change_callback,
+        )
+
+logging.basicConfig(level=logging.DEBUG)
+logg = logging.getLogger()
+
+class StoreFactory:
+
+    def __init__(self, path):
+        self.path = path
+
+
+    def create(self, session_id=None):
+        return SyncRocksDbStore(self.path, session_id=session_id, state_event_callback=state_change_callback, filter_state_event_callback=filter_change_callback)
+
+
+class TestRocksDb(TestStoreBase):
+
+    def setUp(self):
+        super(TestRocksDb, self).setUp()
+        self.store_factory = StoreFactory(self.path).create
+
+
+if __name__ == '__main__':
+    TestStoreBase.link(TestRocksDb)
+    unittest.main()