Compare commits

...

13 Commits

Author SHA1 Message Date
lash 7d5ceb9a28
Bump version 2022-05-10 18:35:13 +00:00
lash 21db658575
Add missing single flag 2022-05-10 18:30:28 +00:00
lash 909b85b2b7
Upgrade deps 2022-05-09 19:43:36 +00:00
lash 2f9663a8f8
Fix faulty new fs store path create 2022-05-07 11:28:13 +00:00
lash 637ead1a38
Improve log output for unlocking tool 2022-05-07 11:27:01 +00:00
lash 412018fc64
Upgrade shep to avoid sync in persist set 2022-05-05 17:05:39 +00:00
lash 95663621bc
Update changelog 2022-05-05 15:43:54 +00:00
lash 0726f7a730
Upgrade shep, handle exception in filestore list in shep 2022-05-05 15:39:18 +00:00
lash 2c8ad85307
Upgrade chainlib, shep (state lock integrity) 2022-05-05 14:49:34 +00:00
lash 5df6656981
Reinstate next tx index in sync state for filter execution 2022-05-05 05:23:06 +00:00
lash 4af735e5d3
Add tests for begin, end, middle revert and continue filter after lock 2022-04-30 07:49:38 +00:00
lash 6b6b26f1ae
WIP Move unlock filter code into store base 2022-04-30 07:35:15 +00:00
lash 44bdda80bf
Correct executable module in cli tool setip 2022-04-29 06:24:00 +00:00
15 changed files with 405 additions and 74 deletions

View File

@ -1,3 +1,22 @@
* 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

View File

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

View File

@ -17,4 +17,6 @@ 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

View File

@ -31,16 +31,25 @@ 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):
# """

View File

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

View File

@ -1 +0,0 @@

View File

@ -106,40 +106,7 @@ def main():
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)
store.unlock_filter(not action_is_forward)
if __name__ == '__main__':

View File

@ -13,12 +13,15 @@ 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()

View File

@ -13,6 +13,7 @@ from chainsyncer.error import (
InterruptError,
IncompleteFilterError,
SyncDone,
FilterInitializationError,
)
logg = logging.getLogger(__name__)
@ -42,7 +43,6 @@ 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,16 +66,6 @@ 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:
@ -110,10 +100,6 @@ 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()
@ -221,9 +207,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)
o.resume()
self.items[offset] = o
self.item_keys.append(offset)
k = str(offset)
self.items[k] = o
self.item_keys.append(k)
elif offset > 0:
logg.warning('block number argument {} for start ignored for already initiated sync {}'.format(offset, self.session_id))
self.started = True
@ -271,12 +257,14 @@ 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)
o.resume()
self.items[block_number] = o
self.item_keys.append(block_number)
k = str(block_number)
self.items[k] = o
self.item_keys.append(k)
logg.info('added existing {}'.format(o))
self.get_target()
v = self.get_target()
if v != None:
target = v
if len(thresholds) == 0:
if self.target != None:
@ -315,8 +303,73 @@ class SyncStore:
raise NotImplementedError()
def peek_next_filter(self):
pass
def __get_locked_item(self):
locked_item = self.filter_state.list(self.filter_state.state_store.LOCK)
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

View File

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

View File

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

View File

@ -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)

View File

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

View File

@ -1,6 +1,6 @@
[metadata]
name = chainsyncer
version = 0.4.1
version = 0.4.10
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.6
python_requires = >= 3.7
packages =
chainsyncer
chainsyncer.driver
@ -37,5 +37,4 @@ packages =
[options.entry_points]
console_scripts =
#blocksync-celery = chainsyncer.runnable.tracker:main
chainsyncer-unlock = chainsyncer.runnable.unlock:main

View File

@ -19,17 +19,22 @@ 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)
self.store = SyncFsStore(self.path, state_event_callback=state_event_handler, filter_state_event_callback=filter_state_event_handler)
self.session = SyncSession(self.store)
self.conn = MockConn()
@ -73,6 +78,279 @@ 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()