CHANGELOG

@@ -1,22 +1,3 @@
-* 0.4.8
-	- Add unlock action description to info loglevel for unlock tool
-* 0.4.7
-	- Upgrade shep to avoid sync in persist set
-* 0.4.6
-	- Upgrade shep to handle filesystem list exception
-* 0.4.5
-	- Upgrade chainlib
-	- Upgrade shep to guarantee atomic state locks
-* 0.4.4
-	- Reinstate tx index bump in sync state on filter execution complete
-* 0.4.3
-	- Refactor lock code, move into module from runnable script
-* 0.4.2
-	- Same as 0.4.1
-* 0.4.1
-	- Correct runnable entry in setup
-* 0.4.0
-	- Refactor on shep
 * 0.3.7
 	- Remove hard eth dependency in settings rendering
 	- Add unlock cli tool

chainsyncer/cli/arg.py

@@ -7,7 +7,6 @@ def process_flags(argparser, flags):
     if flags & SyncFlag.RANGE > 0:
         argparser.add_argument('--offset', type=int, help='Block to start sync from. Default is start of history (0).')
         argparser.add_argument('--until', type=int, default=-1, help='Block to stop sync on. Default is stop at block height of first run.')
-        argparser.add_argument('--single', action='store_true', help='Execute a single sync, regardless of previous states')
     if flags & SyncFlag.HEAD > 0:
         argparser.add_argument('--head', action='store_true', help='Start from latest block as offset')
         argparser.add_argument('--keep-alive', action='store_true', help='Do not stop syncing when caught up')

chainsyncer/cli/config.py

@@ -17,6 +17,4 @@ def process_config(config, args, flags):
             config.add(getattr(args, 'keep_alive'), '_KEEP_ALIVE')
             config.add(getattr(args, 'head'), '_HEAD')
 
-        config.add(getattr(args, 'single'), '_SINGLE')
-
         return config

chainsyncer/error.py

@@ -31,25 +31,16 @@ class LockError(Exception):
 class FilterDone(Exception):
     """Exception raised when all registered filters have been executed
     """
-    pass
 
 
 class InterruptError(FilterDone):
     """Exception for interrupting or attempting to use an interrupted sync
     """
-    pass
 
 
 class IncompleteFilterError(Exception):
     """Exception raised if filter reset is executed prematurely
     """
-    pass
-
-
-class FilterInitializationError(BackendError):
-    """Exception raised if filter state does not match the registered filters
-    """
-    pass
 
 #class AbortTx(Exception):
 #    """

chainsyncer/filter.py

@@ -97,6 +97,7 @@ class FilterState:
             if self.scan != None:
                 ks = self.scan()
                 for v in ks: #os.listdir(self.scan_path):
+                    logg.debug('ks {}'.format(v))
                     k = None
                     try:
                         k = self.state_store.from_elements(v)

chainsyncer/paths.py

@@ -0,0 +1 @@
+

chainsyncer/runnable/unlock.py

@@ -106,7 +106,40 @@ def main():
 
     store.connect()
     store.start(ignore_lock=True)
-    store.unlock_filter(not action_is_forward)
+
+    lock_state = store.filter_state.from_name('LOCK')
+    locked_item = store.filter_state.list(lock_state)
+    if len(locked_item) == 0:
+        sys.stderr.write('Sync filter in {} is not locked\n'.format(state_dir))
+        sys.exit(1)
+    elif len(locked_item) > 1:
+        sys.stderr.write('More than one locked item encountered in {}. That should never happen, so I do not know what to do next.\n'.format(state_dir))
+        sys.exit(1)
+
+    locked_item_key = locked_item[0]
+    locked_item = store.get(int(locked_item_key))
+    locked_state = store.filter_state.state(locked_item_key) - lock_state
+    locked_state_name = store.filter_state.name(locked_state)
+    logg.info('found item "{}" in locked state {}'.format(locked_item, store.filter_state.name(locked_state)))
+
+    if action_is_forward:
+        k = locked_state_name
+        filter_index = None
+        filter_index = filter_list.index(k)
+        filter_pos = filter_index + 1
+        filter_count = len(filter_list)
+        logg.debug('Locked filter {} found at position {} of {}'.format(k, filter_pos, filter_count))
+        if filter_pos == filter_count:
+            logg.info('Locked filter {} is the last filter in the list. Executing filter reset'.format(k))
+            locked_item.reset(check_incomplete=False)
+        else:
+            locked_item.advance(ignore_lock=True)
+            store.filter_state.unset(locked_item_key, lock_state)
+    else:
+        filter_mask = 0xf
+        filter_state = store.filter_state.mask(locked_state, filter_mask)
+        logg.info('Chosen action is "{}": will continue execution at previous filter {}'.format(args.action, store.filter_state.name(filter_state)))
+        store.filter_state.unset(locked_item_key, lock_state)
 
 
 if __name__ == '__main__':

chainsyncer/session.py

@@ -13,15 +13,12 @@ class SyncSession:
     def __init__(self, session_store):
         self.session_store = session_store
         self.started = self.session_store.started
+        self.get = self.session_store.get
         self.next = self.session_store.next_item
         self.item = None
         self.filters = self.session_store.filters
 
     
-    def get(self, k):
-        return self.session_store.get(str(k))
-   
-
     def start(self, offset=0, target=-1):
         self.session_store.start(offset=offset, target=target)
         self.item = self.session_store.next_item()

chainsyncer/store/base.py

@@ -13,7 +13,6 @@ from chainsyncer.error import (
         InterruptError,
         IncompleteFilterError,
         SyncDone,
-        FilterInitializationError,
         )
 
 logg = logging.getLogger(__name__)
@@ -43,6 +42,7 @@ class SyncItem:
         self.filter_state = filter_state
         self.state_key = str(offset)
 
+        logg.debug('get key {}'.format(self.state_key))
         v = self.sync_state.get(self.state_key)
 
         (self.cursor, self.tx_cursor, self.target) = sync_state_deserialize(v)
@@ -66,6 +66,16 @@ class SyncItem:
             raise FilterDone(self.state_key)
 
 
+    def resume(self):
+        filter_state = self.filter_state.state(self.state_key)
+        if filter_state > 0x0f:
+            filter_state_part = self.filter_state.mask(filter_state, 0x0f)
+            if len(self.filter_state.elements(filter_state)) == 1:
+                logg.info('resume execution on state {} ({})'.format(self.filter_state.name(filter_state_part), filter_state_part))
+                lock_state = self.filter_state.from_name('LOCK')
+                self.filter_state.set(lock_state)
+
+
     def reset(self, check_incomplete=True):
         if check_incomplete:
             if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') > 0:
@@ -100,6 +110,10 @@ class SyncItem:
         self.sync_state.replace(self.state_key, b)
 
 
+    def __find_advance(self):
+        v = self.filter_state.state(self.state_key)
+
+
     def advance(self, ignore_lock=False):
         if self.skip_filter:
             raise FilterDone()
@@ -207,9 +221,9 @@ class SyncStore:
             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, ignore_lock=ignore_lock)
-            k = str(offset)
+            o.resume()
-            self.items[k] = o
+            self.items[offset] = o
-            self.item_keys.append(k)
+            self.item_keys.append(offset)
         elif offset > 0:
             logg.warning('block number argument {} for start ignored for already initiated sync {}'.format(offset, self.session_id))
         self.started = True
@@ -257,14 +271,12 @@ class SyncStore:
             if i < lim:
                 item_target = thresholds[i+1] 
             o = SyncItem(block_number, item_target, self.state, self.filter_state, started=True, ignore_lock=ignore_lock)
-            k = str(block_number)
+            o.resume()
-            self.items[k] = o
+            self.items[block_number] = o
-            self.item_keys.append(k)
+            self.item_keys.append(block_number)
             logg.info('added existing {}'.format(o))
 
-        v = self.get_target()
+        self.get_target()
-        if v != None:
-            target = v
         
         if len(thresholds) == 0:
             if self.target != None:
@@ -303,73 +315,8 @@ class SyncStore:
         raise NotImplementedError()
 
 
-    def __get_locked_item(self):
+    def peek_next_filter(self):
-        locked_item = self.filter_state.list(self.filter_state.state_store.LOCK)
+        pass
-        
-        if len(locked_item) == 0:
-            logg.error('Sync filter in store {} is not locked\n'.format(self))
-            return None
-        elif len(locked_item) > 1:
-            raise FilterInitializationError('More than one locked filter item encountered in store {}. That should never happen, so I do not know what to do next.\n'.format(self))
-        return locked_item[0]
-
-
-    def __get_filter_index(self, k):
-        i = -1
-        fltrs = self.load_filter_list()
-        for fltr in fltrs:
-            i += 1
-            if k == fltr.upper():
-                logg.debug('lock filter match at filter list index {}'.format(i))
-        return (i, fltrs,)
-
-
-    def unlock_filter(self, revert=False):
-        locked_item_key = self.__get_locked_item()
-        if locked_item_key == None:
-            return False
-        locked_item = self.get(locked_item_key)
-        state = self.filter_state.state(locked_item_key)
-        locked_state = state - self.filter_state.state_store.LOCK
-        locked_state_name = self.filter_state.name(locked_state)
-
-        logg.debug('found locked item {} in state {}'.format(locked_item, locked_state))
-
-        (i, fltrs) = self.__get_filter_index(locked_state_name)
-
-        if i == -1:
-            raise FilterInitializationError('locked state {} ({}) found for item {}, but matching filter has not been registered'.format(locked_state_name, locked_state, locked_item))
-
-        direction = None
-        if revert:
-            self.__unlock_previous(locked_item, fltrs, i)
-            new_state = self.filter_state.state(locked_item_key)
-            direction = 'previous'
-        else:
-            self.__unlock_next(locked_item, fltrs, i)
-            new_state = self.filter_state.state(locked_item_key)
-            direction = 'next'
-
-        logg.info('chainstate unlock to {} {} ({}) -> {} ({})'.format(direction, self.filter_state.name(state), state, self.filter_state.name(new_state), new_state))
-
-        return True
-
-
-    def __unlock_next(self, item, lst, index):
-        if index == len(lst) - 1:
-            item.reset(check_incomplete=False)
-        else:
-            item.release()
-
-
-    def __unlock_previous(self, item, lst, index):
-        if index == 0:
-            item.reset(check_incomplete=False)
-        else:
-            new_state_str = lst[index - 1]
-            new_state = self.filter_state.state_store.from_name(new_state_str)
-            self.filter_state.state_store.move(item.state_key, new_state)
-
 
     def peek_current_filter(self):
         pass

chainsyncer/store/fs.py

@@ -24,8 +24,7 @@ class SyncFsStore(SyncStore):
             create_path = True
 
         if create_path:
-            #self.__create_path(base_path, self.default_path, session_id=session_id)
+            self.__create_path(base_path, self.default_path, session_id=session_id)
-            os.makedirs(self.session_path)
 
         self.session_id = os.path.basename(self.session_path)
         logg.info('session id {} resolved {} path {}'.format(session_id, self.session_id, self.session_path))

chainsyncer/unittest/base.py

@@ -13,8 +13,7 @@ from chainsyncer.error import NoBlockForYou
 from chainsyncer.driver import SyncDriver
 
 logging.STATETRACE = 5
-logging.addLevelName('STATETRACE', logging.STATETRACE)
+logg = logging.getLogger().getChild(__name__)
-logg = logging.getLogger(__name__)
 
 
 def state_event_handler(k, v_old, v_new):

chainsyncer/unittest/store.py

@@ -232,7 +232,7 @@ class TestStoreBase(unittest.TestCase):
         if self.persist:
             store = self.store_factory('foo')
             store.start()
-            o = store.get('2')
+            o = store.get(2)
 
 
     def t_sync_history_interrupted(self):
@@ -252,7 +252,7 @@ class TestStoreBase(unittest.TestCase):
         store.stop(bogus_item)
         store = self.store_factory('foo')
         store.start()
-        o = store.get('0')
+        o = store.get(0)
         self.assertEqual(o.cursor, 2)
         self.assertEqual(o.target, 13) 
         o.next(advance_block=True)

requirements.txt

@@ -1,5 +1,5 @@
 confini~=0.6.0
 semver==2.13.0
-hexathon~=0.1.6
+hexathon~=0.1.5
-chainlib~=0.2.0
+chainlib~=0.1.1
-shep~=0.2.9
+shep~=0.2.3

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = chainsyncer
-version = 0.4.10
+version = 0.4.1
 description = Generic blockchain syncer driver
 author = Louis Holbrook
 author_email = dev@holbrook.no
@@ -22,7 +22,7 @@ licence_files =
 
 [options]
 include_package_data = True
-python_requires = >= 3.7
+python_requires = >= 3.6
 packages =
 	chainsyncer
 	chainsyncer.driver
@@ -37,4 +37,5 @@ packages =
 
 [options.entry_points]
 console_scripts =
+	#blocksync-celery = chainsyncer.runnable.tracker:main
 	chainsyncer-unlock = chainsyncer.runnable.unlock:main

tests/test_filter.py

@@ -19,22 +19,17 @@ from chainsyncer.unittest import (
         MockConn,
         MockTx,
         MockBlock,
-        MockFilterError,
-        state_event_handler,
-        filter_state_event_handler,
         )
 
-
+logging.basicConfig(level=logging.DEBUG)
-logging.basicConfig(level=logging.STATETRACE)
 logg = logging.getLogger()
 
 
-
 class TestFilter(unittest.TestCase):
 
     def setUp(self):
         self.path = tempfile.mkdtemp()
-        self.store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
+        self.store = SyncFsStore(self.path)
         self.session = SyncSession(self.store)
         self.conn = MockConn()
 
@@ -78,279 +73,6 @@ class TestFilter(unittest.TestCase):
         self.assertEqual(len(fltr_two.contents), 0)
 
 
-    def test_filter_resume_single_revert(self):
-        fltr_one = MockFilter('foo', brk_hard=True)
-        self.store.register(fltr_one)
-
-        self.session.start()
-
-        item = self.store.get('0')
-        item.next()
-
-        tx_hash = os.urandom(32).hex()
-        tx = MockTx(42, tx_hash)
-        block = MockBlock(13, [tx_hash])
-
-        with self.assertRaises(MockFilterError):
-            self.session.filter(self.conn, block, tx)
-      
-        # Unlock the state, reverting to previous filter
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        self.conn = MockConn()
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        store.connect()
-        store.start(ignore_lock=True)
-        store.unlock_filter(revert=True)
-
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        session = SyncSession(store)
-        self.conn = MockConn()
-
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-
-        session.start()
-
-        session.filter(self.conn, block, tx)
-
-
-
-    def test_filter_resume_single_continue(self):
-        fltr_one = MockFilter('foo', brk_hard=True)
-        self.store.register(fltr_one)
-
-        self.session.start()
-
-        item = self.store.get('0')
-        item.next()
-
-        tx_hash = os.urandom(32).hex()
-        tx = MockTx(42, tx_hash)
-        block = MockBlock(13, [tx_hash])
-
-        with self.assertRaises(MockFilterError):
-            self.session.filter(self.conn, block, tx)
-      
-        # Unlock the state, reverting to previous filter
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        self.conn = MockConn()
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        store.connect()
-        store.start(ignore_lock=True)
-        store.unlock_filter(revert=False)
-
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        session = SyncSession(store)
-        self.conn = MockConn()
-
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        store.connect()
-
-        session.start()
-
-        session.filter(self.conn, block, tx)
-
-
-
-    def test_filter_resume_multi_revert_last(self):
-        fltr_one = MockFilter('foo')
-        self.store.register(fltr_one)
-
-        fltr_two = MockFilter('bar', brk_hard=True)
-        self.store.register(fltr_two)
-
-        self.session.start()
-
-        item = self.store.get('0')
-        item.next()
-
-        tx_hash = os.urandom(32).hex()
-        tx = MockTx(42, tx_hash)
-        block = MockBlock(13, [tx_hash])
-
-        with self.assertRaises(MockFilterError):
-            self.session.filter(self.conn, block, tx)
-      
-        # Unlock the state, reverting to previous filter
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        self.conn = MockConn()
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        fltr_bar = MockFilter('bar')
-        store.register(fltr_bar)
-        store.connect()
-        store.start(ignore_lock=True)
-        store.unlock_filter(revert=True)
-
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        session = SyncSession(store)
-        self.conn = MockConn()
-
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        fltr_two = MockFilter('bar')
-        store.register(fltr_two)
-
-        store.connect()
-
-        session.start()
-
-        session.filter(self.conn, block, tx)
-
-
-    def test_filter_resume_multi_continue_last(self):
-        fltr_one = MockFilter('foo')
-        self.store.register(fltr_one)
-
-        fltr_two = MockFilter('bar', brk_hard=True)
-        self.store.register(fltr_two)
-
-        self.session.start()
-
-        item = self.store.get('0')
-        item.next()
-
-        tx_hash = os.urandom(32).hex()
-        tx = MockTx(42, tx_hash)
-        block = MockBlock(13, [tx_hash])
-
-        with self.assertRaises(MockFilterError):
-            self.session.filter(self.conn, block, tx)
-      
-        # Unlock the state, reverting to previous filter
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        self.conn = MockConn()
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        fltr_bar = MockFilter('bar')
-        store.register(fltr_bar)
-        store.connect()
-        store.start(ignore_lock=True)
-        store.unlock_filter(revert=False)
-
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        session = SyncSession(store)
-        self.conn = MockConn()
-
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        fltr_two = MockFilter('bar')
-        store.register(fltr_two)
-
-        session.start()
-
-        session.filter(self.conn, block, tx)
-
-
-    def test_filter_resume_multi_revert_middle(self):
-        fltr_one = MockFilter('foo')
-        self.store.register(fltr_one)
-
-        fltr_two = MockFilter('bar', brk_hard=True)
-        self.store.register(fltr_two)
-
-        fltr_three = MockFilter('baz')
-        self.store.register(fltr_three)
-
-        self.session.start()
-
-        item = self.store.get('0')
-        item.next()
-
-        tx_hash = os.urandom(32).hex()
-        tx = MockTx(42, tx_hash)
-        block = MockBlock(13, [tx_hash])
-
-        with self.assertRaises(MockFilterError):
-            self.session.filter(self.conn, block, tx)
-      
-        # Unlock the state, reverting to previous filter
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        self.conn = MockConn()
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        fltr_two = MockFilter('bar')
-        store.register(fltr_two)
-        fltr_three = MockFilter('baz')
-        store.register(fltr_three)
-
-        store.connect()
-        store.start(ignore_lock=True)
-        store.unlock_filter(revert=True)
-
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        session = SyncSession(store)
-        self.conn = MockConn()
-
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        fltr_two = MockFilter('bar')
-        store.register(fltr_two)
-        fltr_three = MockFilter('baz')
-        store.register(fltr_three)
-
-        store.connect()
-
-        session.start()
-
-        session.filter(self.conn, block, tx)
-
-
-    def test_filter_resume_multi_continue_middle(self):
-        fltr_one = MockFilter('foo')
-        self.store.register(fltr_one)
-
-        fltr_two = MockFilter('bar', brk_hard=True)
-        self.store.register(fltr_two)
-
-        fltr_three = MockFilter('baz')
-        self.store.register(fltr_three)
-
-        self.session.start()
-
-        item = self.store.get('0')
-        item.next()
-
-        tx_hash = os.urandom(32).hex()
-        tx = MockTx(42, tx_hash)
-        block = MockBlock(13, [tx_hash])
-
-        with self.assertRaises(MockFilterError):
-            self.session.filter(self.conn, block, tx)
-      
-        # Unlock the state, reverting to previous filter
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        self.conn = MockConn()
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        fltr_two = MockFilter('bar')
-        store.register(fltr_two)
-        fltr_three = MockFilter('baz')
-        store.register(fltr_three)
-
-        store.connect()
-        store.start(ignore_lock=True)
-        store.unlock_filter(revert=False)
-
-        store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
-        session = SyncSession(store)
-        self.conn = MockConn()
-
-        fltr_one = MockFilter('foo')
-        store.register(fltr_one)
-        fltr_two = MockFilter('bar')
-        store.register(fltr_two)
-        fltr_three = MockFilter('baz')
-        store.register(fltr_three)
-
-        session.start()
-
-        session.filter(self.conn, block, tx)
-
-
 if __name__ == '__main__':
     unittest.main()