Correct runnable package name

chainsyncer/driver/base.py

@@ -141,3 +141,4 @@ class SyncDriver:
 
     def get(self, conn):
         raise NotImplementedError()
+

chainsyncer/filter.py

@@ -43,7 +43,10 @@ class FilterState:
         self.state_store.add('RESET')
 
         self.state = self.state_store.state
+        self.elements = self.state_store.elements
         self.put = self.state_store.put
+        self.mask = self.state_store.mask
+        self.name = self.state_store.name
         self.set = self.state_store.set
         self.next = self.state_store.next
         self.move = self.state_store.move

chainsyncer/runnable/lock.py

@@ -1,77 +0,0 @@
-# SPDX-License-Identifier: GPL-3.0-or-later
-
-# standard imports
-import os
-import logging
-import sys
-import importlib
-
-# external imports
-import chainlib.cli
-from shep.persist import PersistedState
-
-# local imports
-import chainsyncer.cli
-from chainsyncer.settings import ChainsyncerSettings
-from chainsyncer.store import SyncStore
-from chainsyncer.filter import FilterState
-
-logging.basicConfig(level=logging.WARNING)
-logg = logging.getLogger()
-
-
-arg_flags = chainlib.cli.argflag_std_base | chainlib.cli.Flag.CHAIN_SPEC
-argparser = chainlib.cli.ArgumentParser(arg_flags)
-argparser.add_argument('--state-dir', type=str, dest='state_dir', help='State directory')
-argparser.add_positional('action', type=str, help='Action to take on lock. Repeat means re-run the locked filter. Continue means resume execution for next filter.')
-
-sync_flags = chainsyncer.cli.SyncFlag.RANGE | chainsyncer.cli.SyncFlag.HEAD
-chainsyncer.cli.process_flags(argparser, sync_flags)
-
-args = argparser.parse_args()
-
-base_config_dir = chainsyncer.cli.config_dir,
-config = chainlib.cli.Config.from_args(args, arg_flags, base_config_dir=base_config_dir)
-config = chainsyncer.cli.process_config(config, args, sync_flags)
-config.add(args.state_dir, '_STATE_DIR', False)
-logg.debug('config loaded:\n{}'.format(config))
-
-settings = ChainsyncerSettings()
-settings.process_sync_backend(config)
-logg.debug('settings:\n{}'.format(str(settings)))
-
-
-
-def main():
-    if settings.get('SYNCER_BACKEND') == 'mem':
-        raise ValueError('cannot unlock volatile state store')
-
-    if settings.get('SYNCER_BACKEND') == 'fs': 
-        syncer_store_module = importlib.import_module('shep.store.file')
-        syncer_store_class = getattr(syncer_store_module, 'SimpleFileStoreFactory')
-    elif settings.get('SYNCER_BACKEND') == 'rocksdb':
-        syncer_store_module = importlib.import_module('shep.store.rocksdb')
-        syncer_store_class = getattr(syncer_store_module, 'RocksdbStoreFactory')
-    else:
-        raise NotImplementedError('cannot use backend: {}'.format(config.get('SYNCER_BACKEND')))
-       
-    state_dir = config.get('_STATE_DIR')
-
-    factory = syncer_store_class(state_dir)
-    store = SyncStore(state_dir, no_session=True)
-    #base_state = PersistedState(factory.add, 0, check_alias=False)
-    #state = FilterState(base_state, scan=True)
-    store.setup_filter_state(factory=factory)
-    store.connect()
-    store.filter_state.scan()
-    locked_state = store.filter_state.list(store.filter_state.from_name('RESET'))
-    print(locked_state)
-#    if locked_state == None:
-#        sys.stderr.write('state in {} backend "{}" is not locked\n'.format(state_dir, config.get('SYNCER_BACKEND')))
-#        sys.exit(1)
-
-
-
-
-if __name__ == '__main__':
-    main()

chainsyncer/runnable/unlock.py

@@ -0,0 +1,146 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+# standard imports
+import os
+import logging
+import sys
+import importlib
+
+# external imports
+import chainlib.cli
+from shep.persist import PersistedState
+
+# local imports
+import chainsyncer.cli
+from chainsyncer.settings import ChainsyncerSettings
+from chainsyncer.store import SyncStore
+from chainsyncer.filter import (
+        FilterState,
+        SyncFilter,
+        )
+
+logging.basicConfig(level=logging.WARNING)
+logg = logging.getLogger()
+
+valid_fwd = [
+        'fwd',
+        'forward',
+        'next',
+        'continue',
+        ]
+
+valid_rwd = [
+        'rwd',
+        'rewind',
+        'current',
+        'back',
+        'repeat',
+        'replay',
+        ]
+
+action_is_forward = False
+
+arg_flags = chainlib.cli.argflag_std_base | chainlib.cli.Flag.CHAIN_SPEC
+argparser = chainlib.cli.ArgumentParser(arg_flags)
+argparser.add_argument('--state-dir', type=str, dest='state_dir', help='State directory')
+argparser.add_positional('action', type=str, help='Action to take on lock. Repeat means re-run the locked filter. Continue means resume execution for next filter.')
+
+sync_flags = chainsyncer.cli.SyncFlag.RANGE | chainsyncer.cli.SyncFlag.HEAD
+chainsyncer.cli.process_flags(argparser, sync_flags)
+
+args = argparser.parse_args()
+
+if args.action in valid_fwd:
+    action_is_forward = True
+elif args.action not in valid_rwd:
+    sys.stderr.write('action argument must be one of {} or {}\n'.format(valid_rwd, valid_fwd))
+    sys.exit(1)
+
+
+base_config_dir = chainsyncer.cli.config_dir,
+config = chainlib.cli.Config.from_args(args, arg_flags, base_config_dir=base_config_dir)
+config = chainsyncer.cli.process_config(config, args, sync_flags)
+config.add(args.state_dir, '_STATE_DIR', False)
+logg.debug('config loaded:\n{}'.format(config))
+
+settings = ChainsyncerSettings()
+settings.process_sync_backend(config)
+logg.debug('settings:\n{}'.format(str(settings)))
+
+
+class FilterInNameOnly(SyncFilter):
+
+    def __init__(self, k):
+        self.k = k
+
+
+    def common_name(self):
+        return self.k
+
+
+def main():
+    if settings.get('SYNCER_BACKEND') == 'mem':
+        raise ValueError('cannot unlock volatile state store')
+
+    state_dir = config.get('_STATE_DIR')
+
+    if config.get('SYNCER_BACKEND') == 'fs': 
+            syncer_store_module = importlib.import_module('chainsyncer.store.fs')
+            syncer_store_class = getattr(syncer_store_module, 'SyncFsStore')
+    elif config.get('SYNCER_BACKEND') == 'rocksdb':
+        syncer_store_module = importlib.import_module('chainsyncer.store.rocksdb')
+        syncer_store_class = getattr(syncer_store_module, 'SyncRocksDbStore')
+    else:
+        syncer_store_module = importlib.import_module(config.get('SYNCER_BACKEND'))
+        syncer_store_class = getattr(syncer_store_module, 'SyncStore')
+
+    logg.info('using engine {} module {}.{}'.format(config.get('SYNCER_BACKEND'), syncer_store_module.__file__, syncer_store_class.__name__))
+
+    store = syncer_store_class(state_dir)
+    
+    filter_list = store.load_filter_list()
+    for i, k in enumerate(filter_list):
+        fltr = FilterInNameOnly(k)
+        store.register(fltr)
+        filter_list[i] = k.upper()
+
+    store.connect()
+    store.start(ignore_lock=True)
+
+    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__':
+    main()

chainsyncer/store/base.py

@@ -35,7 +35,7 @@ def sync_state_deserialize(b):
 # NOT thread safe
 class SyncItem:
     
-    def __init__(self, offset, target, sync_state, filter_state, started=False, ignore_invalid=False):
+    def __init__(self, offset, target, sync_state, filter_state, started=False, ignore_lock=False):
         self.offset = offset
         self.target = target
         self.sync_state = sync_state
@@ -47,7 +47,8 @@ class SyncItem:
 
         (self.cursor, self.tx_cursor, self.target) = sync_state_deserialize(v)
 
-        if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') and not ignore_invalid:
+        filter_state = self.filter_state.state(self.state_key)
+        if filter_state  & self.filter_state.from_name('LOCK') > 0 and not ignore_lock:
             raise LockError(self.state_key)
 
         self.count = len(self.filter_state.all(pure=True)) - 4
@@ -56,7 +57,7 @@ class SyncItem:
             self.skip_filter = True
         elif not started:
             self.filter_state.move(self.state_key, self.filter_state.from_name('RESET'))
-
+        
 
     def __check_done(self):
         if self.filter_state.state(self.state_key) & self.filter_state.from_name('INTERRUPT') > 0:
@@ -65,11 +66,22 @@ class SyncItem:
             raise FilterDone(self.state_key)
 
 
-    def reset(self):
-        if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') > 0:
-            raise LockError('reset attempt on {} when state locked'.format(self.state_key))
-        if self.filter_state.state(self.state_key) & self.filter_state.from_name('DONE') == 0:
-            raise IncompleteFilterError('reset attempt on {} when incomplete'.format(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:
+                raise LockError('reset attempt on {} when state locked'.format(self.state_key))
+            if self.filter_state.state(self.state_key) & self.filter_state.from_name('DONE') == 0:
+                raise IncompleteFilterError('reset attempt on {} when incomplete'.format(self.state_key))
         self.filter_state.move(self.state_key, self.filter_state.from_name('RESET'))
 
         
@@ -102,13 +114,16 @@ class SyncItem:
         v = self.filter_state.state(self.state_key)
 
 
-    def advance(self):
+    def advance(self, ignore_lock=False):
         if self.skip_filter:
             raise FilterDone()
         self.__check_done()
 
         if self.filter_state.state(self.state_key) & self.filter_state.from_name('LOCK') > 0:
-            raise LockError('advance attempt on {} when state locked'.format(self.state_key))
+            if ignore_lock:
+                self.filter_state.unset(self.state_key, self.filter_state.from_name('LOCK'))
+            else:
+                raise LockError('advance attempt on {} when state locked'.format(self.state_key))
         done = False
         try:
             self.filter_state.next(self.state_key)
@@ -192,18 +207,21 @@ class SyncStore:
         self.filter_state.register(fltr)
 
 
-    def start(self, offset=0, target=-1):
+    def start(self, offset=0, target=-1, ignore_lock=False):
         if self.started:
             return
 
-        self.load(target)
+        self.save_filter_list() 
+        
+        self.load(target, ignore_lock=ignore_lock)
 
         if self.first:
             state_bytes = sync_state_serialize(offset, 0, target)
             block_number_str = str(offset)
             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)
+            o = SyncItem(offset, target, self.state, self.filter_state, ignore_lock=ignore_lock)
+            o.resume()
             self.items[offset] = o
             self.item_keys.append(offset)
         elif offset > 0:
@@ -228,7 +246,7 @@ class SyncStore:
             self.state.put(str(item.cursor), contents=state_bytes)
 
 
-    def load(self, target):
+    def load(self, target, ignore_lock=False):
         self.state.sync(self.state.NEW)
         self.state.sync(self.state.SYNC)
 
@@ -252,7 +270,8 @@ class SyncStore:
             item_target = target
             if i < lim:
                 item_target = thresholds[i+1] 
-            o = SyncItem(block_number, item_target, self.state, self.filter_state, started=True)
+            o = SyncItem(block_number, item_target, self.state, self.filter_state, started=True, ignore_lock=ignore_lock)
+            o.resume()
             self.items[block_number] = o
             self.item_keys.append(block_number)
             logg.info('added existing {}'.format(o))
@@ -269,7 +288,6 @@ class SyncStore:
 
 
     def get(self, k):
-        logg.debug('items {}'.format(self.items.keys()))
         return self.items[k]
 
 
@@ -287,3 +305,18 @@ class SyncStore:
 
     def disconnect(self):
         self.filter_state.disconnect()
+
+
+    def save_filter_list(self):
+        raise NotImplementedError()
+       
+
+    def load_filter_list(self):
+        raise NotImplementedError()
+
+
+    def peek_next_filter(self):
+        pass
+
+    def peek_current_filter(self):
+        pass

chainsyncer/store/fs.py

@@ -74,3 +74,25 @@ class SyncFsStore(SyncStore):
         f.write(str(v))
         f.close()
         self.target = v
+
+
+    def load_filter_list(self):
+        fltr = []
+        fp = os.path.join(self.session_path, 'filter_list')
+        f = open(fp, 'r')
+        while True:
+            v = f.readline()
+            if len(v) == 0:
+                break
+            v = v.rstrip()
+            fltr.append(v) 
+        f.close()
+        return fltr
+
+
+    def save_filter_list(self):
+        fp = os.path.join(self.session_path, 'filter_list')
+        f = open(fp, 'w')
+        for fltr in self.filters:
+            f.write(fltr.common_name() + '\n')
+        f.close()

chainsyncer/store/mem.py

@@ -35,3 +35,11 @@ class SyncMemStore(SyncStore):
         if item != None:
             super(SyncMemStore, self).stop(item)
         logg.info('I am an in-memory only state store. I am shutting down now, so all state will now be discarded.')
+
+
+    def save_filter_list(self):
+        pass
+
+
+    def load_filter_list(self):
+        return []

chainsyncer/store/rocksdb.py

@@ -64,3 +64,16 @@ class SyncRocksDbStore(SyncStore):
         if item != None:
             super(SyncRocksDbStore, self).stop(item)
         self.factory.close()
+
+
+    def save_filter_list(self):
+        fltr = []
+        for v in self.filters:
+            fltr.append(v.common_name())
+        self.target_db.put('filter_list', ','.join(fltr))
+
+
+    def load_filter_list(self):
+        v = self.target_db.get('filter_list')
+        v = v.decode('utf-8')
+        return v.split(',')

chainsyncer/unittest/store.py

@@ -50,6 +50,7 @@ class TestStoreBase(unittest.TestCase):
                 "default",
                 "store_start",
                 "store_resume",
+                "filter_list",
                 "sync_process_nofilter",
                 "sync_process_onefilter",
                 "sync_process_outoforder",
@@ -275,3 +276,26 @@ class TestStoreBase(unittest.TestCase):
         o.next(advance_block=True)
         with self.assertRaises(SyncDone):
             o.next(advance_block=True)
+
+
+    def t_filter_list(self):
+        bogus_item = MockItem(0, 0, 0, 0)
+        store = self.store_factory()
+
+        if store.session_path == None:
+            return
+
+        fltr_one = MockFilter('foo_bar')
+        store.register(fltr_one)
+
+        fltr_two = MockFilter('bar_baz')
+        store.register(fltr_two)
+
+        store.start()
+        store.stop(bogus_item)
+
+        store = self.store_factory()
+        r = store.load_filter_list() 
+
+        self.assertEqual(r[0], 'foo_bar')
+        self.assertEqual(r[1], 'bar_baz')

requirements.txt

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

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = chainsyncer
-version = 0.4.0
+version = 0.4.1
 description = Generic blockchain syncer driver
 author = Louis Holbrook
 author_email = dev@holbrook.no
@@ -29,11 +29,13 @@ packages =
 	chainsyncer.unittest
 	chainsyncer.store
 	chainsyncer.cli
+	chainsyncer.runnable
 
 #[options.package_data]
 #* =
 #	sql/*
 
-#[options.entry_points]
-#console_scripts =
-#	blocksync-celery = chainsyncer.runnable.tracker:main
+[options.entry_points]
+console_scripts =
+	#blocksync-celery = chainsyncer.runnable.tracker:main
+	chainsyncer-unlock = chainsyncer.runnable.unlock:main