Upgrade chainqueue, chainsyncer

This commit is contained in:
lash 2022-04-29 06:26:43 +00:00
parent b52d69c3f9
commit e87ec0cd4c
Signed by: lash
GPG Key ID: 21D2E7BB88C2A746
5 changed files with 31 additions and 7 deletions

View File

@ -58,6 +58,9 @@ class ChaindFsAdapter(ChaindAdapter):
def succeed(self, block, tx): def succeed(self, block, tx):
if self.store.is_reserved(tx.hash):
raise QueueLockError(tx.hash)
return self.store.final(tx.hash, block, tx, error=False) return self.store.final(tx.hash, block, tx, error=False)

View File

@ -16,3 +16,7 @@ class ClientBlockError(BlockingIOError):
class ClientInputError(ValueError): class ClientInputError(ValueError):
pass pass
class QueueLockError(Exception):
pass

View File

@ -1,15 +1,21 @@
# standard imports # standard imports
import logging import logging
import time
# external imports # external imports
from chainlib.status import Status as TxStatus from chainlib.status import Status as TxStatus
from chainsyncer.filter import SyncFilter from chainsyncer.filter import SyncFilter
from chainqueue.error import NotLocalTxError from chainqueue.error import NotLocalTxError
# local imports
from .error import QueueLockError
logg = logging.getLogger(__name__) logg = logging.getLogger(__name__)
class StateFilter(SyncFilter): class StateFilter(SyncFilter):
delay_limit = 3.0
def __init__(self, adapter, throttler=None): def __init__(self, adapter, throttler=None):
self.adapter = adapter self.adapter = adapter
self.throttler = throttler self.throttler = throttler
@ -22,10 +28,21 @@ class StateFilter(SyncFilter):
logg.debug('skipping not local transaction {}'.format(tx.hash)) logg.debug('skipping not local transaction {}'.format(tx.hash))
return False return False
if tx.status == TxStatus.SUCCESS: delay = 0.01
self.adapter.succeed(block, tx) while True:
else: if delay > self.delay_limit:
self.adapter.fail(block, tx) raise QueueLockError('The queue lock for tx {} seems to be stuck. Human meddling needed.'.format(tx.hash))
try:
if tx.status == TxStatus.SUCCESS:
self.adapter.succeed(block, tx)
else:
self.adapter.fail(block, tx)
break
except QueueLockError as e:
logg.debug('queue item {} is blocked, will retry: {}'.format(tx.hash, e))
time.sleep(delay)
delay *= 2
if self.throttler != None: if self.throttler != None:
self.throttler.dec(tx.hash) self.throttler.dec(tx.hash)

View File

@ -1,6 +1,6 @@
chainlib~=0.1.1 chainlib~=0.1.1
chainqueue~=0.1.2 chainqueue~=0.1.5
chainsyncer~=0.4.0 chainsyncer~=0.4.2
confini~=0.6.0 confini~=0.6.0
funga~=0.5.2 funga~=0.5.2
pyxdg~=0.26 pyxdg~=0.26

View File

@ -1,6 +1,6 @@
[metadata] [metadata]
name = chaind name = chaind
version = 0.1.2 version = 0.1.3
description = Base package for chain queue service description = Base package for chain queue service
author = Louis Holbrook author = Louis Holbrook
author_email = dev@holbrook.no author_email = dev@holbrook.no