Add chainsyncer db migration, use chainsyncer in cluster, rename manager to tracker
This commit is contained in:
parent
dfd0de32e1
commit
a766a9de70
@ -0,0 +1,28 @@
|
|||||||
|
"""Add chain syncer
|
||||||
|
|
||||||
|
Revision ID: ec40ac0974c1
|
||||||
|
Revises: 6ac7a1dadc46
|
||||||
|
Create Date: 2021-02-23 06:10:19.246304
|
||||||
|
|
||||||
|
"""
|
||||||
|
from alembic import op
|
||||||
|
import sqlalchemy as sa
|
||||||
|
from chainsyncer.db.migrations.sqlalchemy import (
|
||||||
|
chainsyncer_upgrade,
|
||||||
|
chainsyncer_downgrade,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# revision identifiers, used by Alembic.
|
||||||
|
revision = 'ec40ac0974c1'
|
||||||
|
down_revision = '6ac7a1dadc46'
|
||||||
|
branch_labels = None
|
||||||
|
depends_on = None
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade():
|
||||||
|
chainsyncer_upgrade(0, 0, 1)
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade():
|
||||||
|
chainsyncer_downgrade(0, 0, 1)
|
@ -0,0 +1,28 @@
|
|||||||
|
"""Add chain syncer
|
||||||
|
|
||||||
|
Revision ID: ec40ac0974c1
|
||||||
|
Revises: 6ac7a1dadc46
|
||||||
|
Create Date: 2021-02-23 06:10:19.246304
|
||||||
|
|
||||||
|
"""
|
||||||
|
from alembic import op
|
||||||
|
import sqlalchemy as sa
|
||||||
|
from chainsyncer.db.migrations.sqlalchemy import (
|
||||||
|
chainsyncer_upgrade,
|
||||||
|
chainsyncer_downgrade,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# revision identifiers, used by Alembic.
|
||||||
|
revision = 'ec40ac0974c1'
|
||||||
|
down_revision = '6ac7a1dadc46'
|
||||||
|
branch_labels = None
|
||||||
|
depends_on = None
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade():
|
||||||
|
chainsyncer_upgrade(0, 0, 1)
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade():
|
||||||
|
chainsyncer_downgrade(0, 0, 1)
|
@ -112,5 +112,6 @@ class CallbackFilter(SyncFilter):
|
|||||||
except UnknownContractError:
|
except UnknownContractError:
|
||||||
logg.debug('callback filter {}:{} skipping "transfer" method on unknown contract {} tx {}'.format(tc.queue, tc.method, transfer_data['to'], tx.hash.hex()))
|
logg.debug('callback filter {}:{} skipping "transfer" method on unknown contract {} tx {}'.format(tc.queue, tc.method, transfer_data['to'], tx.hash.hex()))
|
||||||
|
|
||||||
|
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
return 'cic-eth callbacks'
|
return 'cic-eth callbacks'
|
||||||
|
1
apps/cic-eth/cic_eth/sync/__init__.py
Normal file
1
apps/cic-eth/cic_eth/sync/__init__.py
Normal file
@ -0,0 +1 @@
|
|||||||
|
from .base import Syncer
|
201
apps/cic-eth/cic_eth/sync/backend.py
Normal file
201
apps/cic-eth/cic_eth/sync/backend.py
Normal file
@ -0,0 +1,201 @@
|
|||||||
|
# standard imports
|
||||||
|
import logging
|
||||||
|
|
||||||
|
# local imports
|
||||||
|
from cic_eth.db.models.sync import BlockchainSync
|
||||||
|
from cic_eth.db.models.base import SessionBase
|
||||||
|
|
||||||
|
logg = logging.getLogger()
|
||||||
|
|
||||||
|
|
||||||
|
class SyncerBackend:
|
||||||
|
"""Interface to block and transaction sync state.
|
||||||
|
|
||||||
|
:param chain_spec: Chain spec for the chain that syncer is running for.
|
||||||
|
:type chain_spec: cic_registry.chain.ChainSpec
|
||||||
|
:param object_id: Unique id for the syncer session.
|
||||||
|
:type object_id: number
|
||||||
|
"""
|
||||||
|
def __init__(self, chain_spec, object_id):
|
||||||
|
self.db_session = None
|
||||||
|
self.db_object = None
|
||||||
|
self.chain_spec = chain_spec
|
||||||
|
self.object_id = object_id
|
||||||
|
self.connect()
|
||||||
|
self.disconnect()
|
||||||
|
|
||||||
|
|
||||||
|
def connect(self):
|
||||||
|
"""Loads the state of the syncer session with the given id.
|
||||||
|
"""
|
||||||
|
if self.db_session == None:
|
||||||
|
self.db_session = SessionBase.create_session()
|
||||||
|
q = self.db_session.query(BlockchainSync)
|
||||||
|
q = q.filter(BlockchainSync.id==self.object_id)
|
||||||
|
self.db_object = q.first()
|
||||||
|
if self.db_object == None:
|
||||||
|
self.disconnect()
|
||||||
|
raise ValueError('sync entry with id {} not found'.format(self.object_id))
|
||||||
|
return self.db_session
|
||||||
|
|
||||||
|
|
||||||
|
def disconnect(self):
|
||||||
|
"""Commits state of sync to backend.
|
||||||
|
"""
|
||||||
|
if self.db_session != None:
|
||||||
|
self.db_session.add(self.db_object)
|
||||||
|
self.db_session.commit()
|
||||||
|
self.db_session.close()
|
||||||
|
self.db_session = None
|
||||||
|
|
||||||
|
|
||||||
|
def chain(self):
|
||||||
|
"""Returns chain spec for syncer
|
||||||
|
|
||||||
|
:returns: Chain spec
|
||||||
|
:rtype chain_spec: cic_registry.chain.ChainSpec
|
||||||
|
"""
|
||||||
|
return self.chain_spec
|
||||||
|
|
||||||
|
|
||||||
|
def get(self):
|
||||||
|
"""Get the current state of the syncer cursor.
|
||||||
|
|
||||||
|
:returns: Block and block transaction height, respectively
|
||||||
|
:rtype: tuple
|
||||||
|
"""
|
||||||
|
self.connect()
|
||||||
|
pair = self.db_object.cursor()
|
||||||
|
self.disconnect()
|
||||||
|
return pair
|
||||||
|
|
||||||
|
|
||||||
|
def set(self, block_height, tx_height):
|
||||||
|
"""Update the state of the syncer cursor
|
||||||
|
:param block_height: Block height of cursor
|
||||||
|
:type block_height: number
|
||||||
|
:param tx_height: Block transaction height of cursor
|
||||||
|
:type tx_height: number
|
||||||
|
:returns: Block and block transaction height, respectively
|
||||||
|
:rtype: tuple
|
||||||
|
"""
|
||||||
|
self.connect()
|
||||||
|
pair = self.db_object.set(block_height, tx_height)
|
||||||
|
self.disconnect()
|
||||||
|
return pair
|
||||||
|
|
||||||
|
|
||||||
|
def start(self):
|
||||||
|
"""Get the initial state of the syncer cursor.
|
||||||
|
|
||||||
|
:returns: Initial block and block transaction height, respectively
|
||||||
|
:rtype: tuple
|
||||||
|
"""
|
||||||
|
self.connect()
|
||||||
|
pair = self.db_object.start()
|
||||||
|
self.disconnect()
|
||||||
|
return pair
|
||||||
|
|
||||||
|
|
||||||
|
def target(self):
|
||||||
|
"""Get the target state (upper bound of sync) of the syncer cursor.
|
||||||
|
|
||||||
|
:returns: Target block height
|
||||||
|
:rtype: number
|
||||||
|
"""
|
||||||
|
self.connect()
|
||||||
|
target = self.db_object.target()
|
||||||
|
self.disconnect()
|
||||||
|
return target
|
||||||
|
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def first(chain):
|
||||||
|
"""Returns the model object of the most recent syncer in backend.
|
||||||
|
|
||||||
|
:param chain: Chain spec of chain that syncer is running for.
|
||||||
|
:type chain: cic_registry.chain.ChainSpec
|
||||||
|
:returns: Last syncer object
|
||||||
|
:rtype: cic_eth.db.models.BlockchainSync
|
||||||
|
"""
|
||||||
|
return BlockchainSync.first(chain)
|
||||||
|
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def initial(chain, block_height):
|
||||||
|
"""Creates a new syncer session and commit its initial state to backend.
|
||||||
|
|
||||||
|
:param chain: Chain spec of chain that syncer is running for.
|
||||||
|
:type chain: cic_registry.chain.ChainSpec
|
||||||
|
:param block_height: Target block height
|
||||||
|
:type block_height: number
|
||||||
|
:returns: New syncer object
|
||||||
|
:rtype: cic_eth.db.models.BlockchainSync
|
||||||
|
"""
|
||||||
|
object_id = None
|
||||||
|
session = SessionBase.create_session()
|
||||||
|
o = BlockchainSync(chain, 0, 0, block_height)
|
||||||
|
session.add(o)
|
||||||
|
session.commit()
|
||||||
|
object_id = o.id
|
||||||
|
session.close()
|
||||||
|
|
||||||
|
return SyncerBackend(chain, object_id)
|
||||||
|
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def resume(chain, block_height):
|
||||||
|
"""Retrieves and returns all previously unfinished syncer sessions.
|
||||||
|
|
||||||
|
|
||||||
|
:param chain: Chain spec of chain that syncer is running for.
|
||||||
|
:type chain: cic_registry.chain.ChainSpec
|
||||||
|
:param block_height: Target block height
|
||||||
|
:type block_height: number
|
||||||
|
:returns: Syncer objects of unfinished syncs
|
||||||
|
:rtype: list of cic_eth.db.models.BlockchainSync
|
||||||
|
"""
|
||||||
|
syncers = []
|
||||||
|
|
||||||
|
session = SessionBase.create_session()
|
||||||
|
|
||||||
|
object_id = None
|
||||||
|
|
||||||
|
for object_id in BlockchainSync.get_unsynced(session=session):
|
||||||
|
logg.debug('block syncer resume added previously unsynced sync entry id {}'.format(object_id))
|
||||||
|
syncers.append(SyncerBackend(chain, object_id))
|
||||||
|
|
||||||
|
(block_resume, tx_resume) = BlockchainSync.get_last_live_height(block_height, session=session)
|
||||||
|
if block_height != block_resume:
|
||||||
|
o = BlockchainSync(chain, block_resume, tx_resume, block_height)
|
||||||
|
session.add(o)
|
||||||
|
session.commit()
|
||||||
|
object_id = o.id
|
||||||
|
syncers.append(SyncerBackend(chain, object_id))
|
||||||
|
logg.debug('block syncer resume added new sync entry from previous run id {}, start{}:{} target {}'.format(object_id, block_resume, tx_resume, block_height))
|
||||||
|
|
||||||
|
session.close()
|
||||||
|
|
||||||
|
return syncers
|
||||||
|
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def live(chain, block_height):
|
||||||
|
"""Creates a new open-ended syncer session starting at the given block height.
|
||||||
|
|
||||||
|
:param chain: Chain spec of chain that syncer is running for.
|
||||||
|
:type chain: cic_registry.chain.ChainSpec
|
||||||
|
:param block_height: Target block height
|
||||||
|
:type block_height: number
|
||||||
|
:returns: "Live" syncer object
|
||||||
|
:rtype: cic_eth.db.models.BlockchainSync
|
||||||
|
"""
|
||||||
|
object_id = None
|
||||||
|
session = SessionBase.create_session()
|
||||||
|
o = BlockchainSync(chain, block_height, 0, None)
|
||||||
|
session.add(o)
|
||||||
|
session.commit()
|
||||||
|
object_id = o.id
|
||||||
|
session.close()
|
||||||
|
|
||||||
|
return SyncerBackend(chain, object_id)
|
51
apps/cic-eth/cic_eth/sync/base.py
Normal file
51
apps/cic-eth/cic_eth/sync/base.py
Normal file
@ -0,0 +1,51 @@
|
|||||||
|
# TODO: extend blocksync model
|
||||||
|
class Syncer:
|
||||||
|
"""Base class and interface for implementing a block sync poller routine.
|
||||||
|
|
||||||
|
:param bc_cache: Retrieves block cache cursors for chain head and latest processed block.
|
||||||
|
:type bc_cache: cic_eth.sync.SyncerBackend
|
||||||
|
"""
|
||||||
|
w3 = None
|
||||||
|
running_global = True
|
||||||
|
|
||||||
|
def __init__(self, bc_cache):
|
||||||
|
self.cursor = None
|
||||||
|
self.bc_cache = bc_cache
|
||||||
|
self.filter = []
|
||||||
|
self.running = True
|
||||||
|
|
||||||
|
|
||||||
|
def chain(self):
|
||||||
|
"""Returns the string representation of the chain spec for the chain the syncer is running on.
|
||||||
|
|
||||||
|
:returns: Chain spec string
|
||||||
|
:rtype: str
|
||||||
|
"""
|
||||||
|
return self.bc_cache.chain()
|
||||||
|
|
||||||
|
|
||||||
|
def get(self):
|
||||||
|
"""Get latest unprocessed blocks.
|
||||||
|
|
||||||
|
:returns: list of block hash strings
|
||||||
|
:rtype: list
|
||||||
|
"""
|
||||||
|
raise NotImplementedError()
|
||||||
|
|
||||||
|
|
||||||
|
def process(self, w3, ref):
|
||||||
|
"""Process transactions in a single block.
|
||||||
|
|
||||||
|
:param ref: Reference of object to process
|
||||||
|
:type ref: str, 0x-hex
|
||||||
|
"""
|
||||||
|
raise NotImplementedError()
|
||||||
|
|
||||||
|
|
||||||
|
def loop(self, interval):
|
||||||
|
"""Entry point for syncer loop
|
||||||
|
|
||||||
|
:param interval: Delay in seconds until next attempt if no new blocks are found.
|
||||||
|
:type interval: int
|
||||||
|
"""
|
||||||
|
raise NotImplementedError()
|
4
apps/cic-eth/cic_eth/sync/error.py
Normal file
4
apps/cic-eth/cic_eth/sync/error.py
Normal file
@ -0,0 +1,4 @@
|
|||||||
|
class LoopDone(Exception):
|
||||||
|
"""Exception raised when a syncing is complete.
|
||||||
|
"""
|
||||||
|
pass
|
51
apps/cic-eth/cic_eth/sync/head.py
Normal file
51
apps/cic-eth/cic_eth/sync/head.py
Normal file
@ -0,0 +1,51 @@
|
|||||||
|
# standard imports
|
||||||
|
import logging
|
||||||
|
|
||||||
|
# third-party imports
|
||||||
|
import web3
|
||||||
|
|
||||||
|
# local imports
|
||||||
|
from .mined import MinedSyncer
|
||||||
|
from .base import Syncer
|
||||||
|
|
||||||
|
logg = logging.getLogger()
|
||||||
|
|
||||||
|
|
||||||
|
class HeadSyncer(MinedSyncer):
|
||||||
|
"""Implements the get method in Syncer for retrieving every new mined block.
|
||||||
|
|
||||||
|
:param bc_cache: Retrieves block cache cursors for chain head and latest processed block.
|
||||||
|
:type bc_cache: Object implementing methods from cic_eth.sync.SyncerBackend
|
||||||
|
"""
|
||||||
|
def __init__(self, bc_cache):
|
||||||
|
super(HeadSyncer, self).__init__(bc_cache)
|
||||||
|
# TODO: filter not returning all blocks, at least with ganache. kind of defeats the point, then
|
||||||
|
#self.w3_filter = rpc.w3.eth.filter({
|
||||||
|
# 'fromBlock': block_offset,
|
||||||
|
# }) #'latest')
|
||||||
|
#self.bc_cache.set(block_offset, 0)
|
||||||
|
logg.debug('initialized head syncer with offset {}'.format(bc_cache.start()))
|
||||||
|
|
||||||
|
"""Implements Syncer.get
|
||||||
|
|
||||||
|
:param w3: Web3 object
|
||||||
|
:type w3: web3.Web3
|
||||||
|
:returns: Block hash of newly mined blocks. if any
|
||||||
|
:rtype: list of str, 0x-hex
|
||||||
|
"""
|
||||||
|
def get(self, w3):
|
||||||
|
# Of course, the filter doesn't return the same block dict format as getBlock() so we'll just waste some cycles getting the hashes instead.
|
||||||
|
#hashes = []
|
||||||
|
#for block in self.w3_filter.get_new_entries():
|
||||||
|
# hashes.append(block['blockHash'])
|
||||||
|
#logg.debug('blocks {}'.format(hashes))
|
||||||
|
#return hashes
|
||||||
|
(block_number, tx_number) = self.bc_cache.get()
|
||||||
|
block_hash = []
|
||||||
|
try:
|
||||||
|
block = w3.eth.getBlock(block_number)
|
||||||
|
block_hash.append(block.hash)
|
||||||
|
except web3.exceptions.BlockNotFound:
|
||||||
|
pass
|
||||||
|
|
||||||
|
return block_hash
|
74
apps/cic-eth/cic_eth/sync/history.py
Normal file
74
apps/cic-eth/cic_eth/sync/history.py
Normal file
@ -0,0 +1,74 @@
|
|||||||
|
# standard imports
|
||||||
|
import logging
|
||||||
|
|
||||||
|
# third-party imports
|
||||||
|
from web3.exceptions import BlockNotFound
|
||||||
|
from .error import LoopDone
|
||||||
|
|
||||||
|
# local imports
|
||||||
|
from .mined import MinedSyncer
|
||||||
|
from .base import Syncer
|
||||||
|
from cic_eth.db.models.base import SessionBase
|
||||||
|
|
||||||
|
logg = logging.getLogger()
|
||||||
|
|
||||||
|
|
||||||
|
class HistorySyncer(MinedSyncer):
|
||||||
|
"""Implements the get method in Syncer for retrieving all blocks between last processed block before previous shutdown and block height at time of syncer start.
|
||||||
|
|
||||||
|
:param bc_cache: Retrieves block cache cursors for chain head and latest processed block.
|
||||||
|
:type bc_cache: Object implementing methods from cic_eth.sync.SyncerBackend
|
||||||
|
:param mx: Maximum number of blocks to return in one call
|
||||||
|
:type mx: int
|
||||||
|
"""
|
||||||
|
def __init__(self, bc_cache, mx=500):
|
||||||
|
super(HistorySyncer, self).__init__(bc_cache)
|
||||||
|
self.max = mx
|
||||||
|
|
||||||
|
self.target = bc_cache.target()
|
||||||
|
logg.info('History syncer target block number {}'.format(self.target))
|
||||||
|
|
||||||
|
session_offset = self.bc_cache.get()
|
||||||
|
|
||||||
|
self.block_offset = session_offset[0]
|
||||||
|
self.tx_offset = session_offset[1]
|
||||||
|
logg.info('History syncer starting at {}:{}'.format(session_offset[0], session_offset[1]))
|
||||||
|
|
||||||
|
self.filter = []
|
||||||
|
|
||||||
|
|
||||||
|
"""Implements Syncer.get
|
||||||
|
|
||||||
|
BUG: Should also raise LoopDone when block array is empty after loop.
|
||||||
|
|
||||||
|
:param w3: Web3 object
|
||||||
|
:type w3: web3.Web3
|
||||||
|
:raises LoopDone: If a block is not found.
|
||||||
|
:return: Return a batch of blocks to process
|
||||||
|
:rtype: list of str, 0x-hex
|
||||||
|
"""
|
||||||
|
def get(self, w3):
|
||||||
|
sync_db = self.bc_cache
|
||||||
|
height = self.bc_cache.get()
|
||||||
|
logg.debug('height {}'.format(height))
|
||||||
|
block_last = height[0]
|
||||||
|
tx_last = height[1]
|
||||||
|
if not self.running:
|
||||||
|
raise LoopDone((block_last, tx_last))
|
||||||
|
b = []
|
||||||
|
block_target = block_last + self.max
|
||||||
|
if block_target > self.target:
|
||||||
|
block_target = self.target
|
||||||
|
logg.debug('target {} last {} max {}'.format(block_target, block_last, self.max))
|
||||||
|
for i in range(block_last, block_target):
|
||||||
|
if i == self.target:
|
||||||
|
logg.info('reached target {}, exiting'.format(i))
|
||||||
|
self.running = False
|
||||||
|
break
|
||||||
|
bhash = w3.eth.getBlock(i).hash
|
||||||
|
b.append(bhash)
|
||||||
|
logg.debug('appending block {} {}'.format(i, bhash.hex()))
|
||||||
|
if block_last == block_target:
|
||||||
|
logg.info('aleady reached target {}, exiting'.format(self.target))
|
||||||
|
self.running = False
|
||||||
|
return b
|
50
apps/cic-eth/cic_eth/sync/mempool.py
Normal file
50
apps/cic-eth/cic_eth/sync/mempool.py
Normal file
@ -0,0 +1,50 @@
|
|||||||
|
class MemPoolSyncer(Syncer):
|
||||||
|
|
||||||
|
|
||||||
|
def __init__(self, bc_cache):
|
||||||
|
raise NotImplementedError('incomplete, needs web3 tx to raw transaction conversion')
|
||||||
|
super(MemPoolSyncer, self).__init__(bc_cache)
|
||||||
|
# self.w3_filter = Syncer.w3.eth.filter('pending')
|
||||||
|
# for tx in tx_cache.txs:
|
||||||
|
# self.txs.append(tx)
|
||||||
|
# logg.debug('add tx {} to mempoolsyncer'.format(tx))
|
||||||
|
#
|
||||||
|
#
|
||||||
|
# def get(self):
|
||||||
|
# return self.w3_filter.get_new_entries()
|
||||||
|
#
|
||||||
|
#
|
||||||
|
# def process(self, tx_hash):
|
||||||
|
# tx_hash_hex = tx_hash.hex()
|
||||||
|
# if tx_hash_hex in self.txs:
|
||||||
|
# logg.debug('syncer already watching {}, skipping'.format(tx_hash_hex))
|
||||||
|
# tx = self.w3.eth.getTransaction(tx_hash_hex)
|
||||||
|
# serialized_tx = rlp.encode({
|
||||||
|
# 'nonce': tx.nonce,
|
||||||
|
# 'from': getattr(tx, 'from'),
|
||||||
|
# })
|
||||||
|
# logg.info('add {} to syncer: {}'.format(tx, serialized_tx))
|
||||||
|
# otx = Otx(
|
||||||
|
# nonce=tx.nonce,
|
||||||
|
# address=getattr(tx, 'from'),
|
||||||
|
# tx_hash=tx_hash_hex,
|
||||||
|
# signed_tx=serialized_tx,
|
||||||
|
# )
|
||||||
|
# Otx.session.add(otx)
|
||||||
|
# Otx.session.commit()
|
||||||
|
#
|
||||||
|
#
|
||||||
|
# def loop(self, interval):
|
||||||
|
# while Syncer.running:
|
||||||
|
# logg.debug('loop execute')
|
||||||
|
# txs = self.get()
|
||||||
|
# logg.debug('got txs {}'.format(txs))
|
||||||
|
# for tx in txs:
|
||||||
|
# #block_number = self.process(block.hex())
|
||||||
|
# self.process(tx)
|
||||||
|
# #if block_number > self.bc_cache.head():
|
||||||
|
# # self.bc_cache.head(block_number)
|
||||||
|
# time.sleep(interval)
|
||||||
|
# logg.info("Syncer no longer set to run, gracefully exiting")
|
||||||
|
|
||||||
|
|
109
apps/cic-eth/cic_eth/sync/mined.py
Normal file
109
apps/cic-eth/cic_eth/sync/mined.py
Normal file
@ -0,0 +1,109 @@
|
|||||||
|
# standard imports
|
||||||
|
import logging
|
||||||
|
import time
|
||||||
|
|
||||||
|
# third-party imports
|
||||||
|
import celery
|
||||||
|
|
||||||
|
# local impotes
|
||||||
|
from .base import Syncer
|
||||||
|
from cic_eth.queue.tx import set_final_status
|
||||||
|
from cic_eth.eth import RpcClient
|
||||||
|
|
||||||
|
app = celery.current_app
|
||||||
|
logg = logging.getLogger()
|
||||||
|
|
||||||
|
|
||||||
|
class MinedSyncer(Syncer):
|
||||||
|
"""Base implementation of block processor for mined blocks.
|
||||||
|
|
||||||
|
Loops through all transactions,
|
||||||
|
|
||||||
|
:param bc_cache: Retrieves block cache cursors for chain head and latest processed block.
|
||||||
|
:type bc_cache: Object implementing methods from cic_eth.sync.SyncerBackend
|
||||||
|
"""
|
||||||
|
|
||||||
|
yield_delay = 0.005
|
||||||
|
|
||||||
|
def __init__(self, bc_cache):
|
||||||
|
super(MinedSyncer, self).__init__(bc_cache)
|
||||||
|
self.block_offset = 0
|
||||||
|
self.tx_offset = 0
|
||||||
|
|
||||||
|
|
||||||
|
def process(self, w3, ref):
|
||||||
|
"""Processes transactions in a single block, advancing transaction (and block) cursor accordingly.
|
||||||
|
|
||||||
|
:param w3: Web3 object
|
||||||
|
:type w3: web3.Web3
|
||||||
|
:param ref: Block reference (hash) to process
|
||||||
|
:type ref: str, 0x-hex
|
||||||
|
:returns: Block number of next unprocessed block
|
||||||
|
:rtype: number
|
||||||
|
"""
|
||||||
|
b = w3.eth.getBlock(ref)
|
||||||
|
c = w3.eth.getBlockTransactionCount(ref)
|
||||||
|
s = 0
|
||||||
|
if self.block_offset == b.number:
|
||||||
|
s = self.tx_offset
|
||||||
|
|
||||||
|
logg.debug('processing {} (blocknumber {}, count {}, offset {})'.format(ref, b.number, c, s))
|
||||||
|
|
||||||
|
for i in range(s, c):
|
||||||
|
tx = w3.eth.getTransactionByBlock(ref, i)
|
||||||
|
tx_hash_hex = tx['hash'].hex()
|
||||||
|
rcpt = w3.eth.getTransactionReceipt(tx_hash_hex)
|
||||||
|
logg.debug('{}/{} processing tx {} from block {} {}'.format(i+1, c, tx_hash_hex, b.number, ref))
|
||||||
|
ours = False
|
||||||
|
# TODO: ensure filter loop can complete on graceful shutdown
|
||||||
|
for f in self.filter:
|
||||||
|
#try:
|
||||||
|
session = self.bc_cache.connect()
|
||||||
|
task_uuid = f(w3, tx, rcpt, self.chain(), session)
|
||||||
|
#except Exception as e:
|
||||||
|
# logg.error('error in filter {} tx {}: {}'.format(f, tx_hash_hex, e))
|
||||||
|
# continue
|
||||||
|
if task_uuid != None:
|
||||||
|
logg.debug('tx {} passed to celery task {}'.format(tx_hash_hex, task_uuid))
|
||||||
|
s = celery.signature(
|
||||||
|
'set_final_status',
|
||||||
|
[tx_hash_hex, rcpt['blockNumber'], not rcpt['status']],
|
||||||
|
)
|
||||||
|
s.apply_async()
|
||||||
|
break
|
||||||
|
next_tx = i + 1
|
||||||
|
if next_tx == c:
|
||||||
|
self.bc_cache.set(b.number+1, 0)
|
||||||
|
else:
|
||||||
|
self.bc_cache.set(b.number, next_tx)
|
||||||
|
if c == 0:
|
||||||
|
logg.info('synced block {} has no transactions'.format(b.number))
|
||||||
|
#self.bc_cache.session(b.number+1, 0)
|
||||||
|
self.bc_cache.set(b.number+1, 0)
|
||||||
|
return b['number']
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
def loop(self, interval):
|
||||||
|
"""Loop running until the "running" property of Syncer is set to False.
|
||||||
|
|
||||||
|
Retrieves latest unprocessed blocks and processes them.
|
||||||
|
|
||||||
|
:param interval: Delay in seconds until next attempt if no new blocks are found.
|
||||||
|
:type interval: int
|
||||||
|
"""
|
||||||
|
while self.running and Syncer.running_global:
|
||||||
|
self.bc_cache.connect()
|
||||||
|
c = RpcClient(self.chain())
|
||||||
|
logg.debug('loop execute')
|
||||||
|
e = self.get(c.w3)
|
||||||
|
logg.debug('got blocks {}'.format(e))
|
||||||
|
for block in e:
|
||||||
|
block_number = self.process(c.w3, block.hex())
|
||||||
|
logg.info('processed block {} {}'.format(block_number, block.hex()))
|
||||||
|
self.bc_cache.disconnect()
|
||||||
|
if len(e) > 0:
|
||||||
|
time.sleep(self.yield_delay)
|
||||||
|
else:
|
||||||
|
time.sleep(interval)
|
||||||
|
logg.info("Syncer no longer set to run, gracefully exiting")
|
71
apps/cic-eth/cic_eth/sync/retry.py
Normal file
71
apps/cic-eth/cic_eth/sync/retry.py
Normal file
@ -0,0 +1,71 @@
|
|||||||
|
# standard imports
|
||||||
|
import logging
|
||||||
|
import datetime
|
||||||
|
import time
|
||||||
|
|
||||||
|
# third-party imports
|
||||||
|
import celery
|
||||||
|
|
||||||
|
# local imports
|
||||||
|
from .base import Syncer
|
||||||
|
from cic_eth.eth.rpc import RpcClient
|
||||||
|
from cic_eth.db.enum import StatusEnum
|
||||||
|
from cic_eth.queue.tx import get_status_tx
|
||||||
|
|
||||||
|
logg = logging.getLogger()
|
||||||
|
|
||||||
|
celery_app = celery.current_app
|
||||||
|
|
||||||
|
|
||||||
|
class noop_cache:
|
||||||
|
|
||||||
|
def __init__(self, chain_spec):
|
||||||
|
self.chain_spec = chain_spec
|
||||||
|
|
||||||
|
|
||||||
|
def chain(self):
|
||||||
|
return self.chain_spec
|
||||||
|
|
||||||
|
|
||||||
|
class RetrySyncer(Syncer):
|
||||||
|
|
||||||
|
def __init__(self, chain_spec, stalled_grace_seconds, failed_grace_seconds=None, final_func=None):
|
||||||
|
cache = noop_cache(chain_spec)
|
||||||
|
super(RetrySyncer, self).__init__(cache)
|
||||||
|
if failed_grace_seconds == None:
|
||||||
|
failed_grace_seconds = stalled_grace_seconds
|
||||||
|
self.stalled_grace_seconds = stalled_grace_seconds
|
||||||
|
self.failed_grace_seconds = failed_grace_seconds
|
||||||
|
self.final_func = final_func
|
||||||
|
|
||||||
|
|
||||||
|
def get(self, w3):
|
||||||
|
# before = datetime.datetime.utcnow() - datetime.timedelta(seconds=self.failed_grace_seconds)
|
||||||
|
# failed_txs = get_status_tx(
|
||||||
|
# StatusEnum.SENDFAIL.value,
|
||||||
|
# before=before,
|
||||||
|
# )
|
||||||
|
before = datetime.datetime.utcnow() - datetime.timedelta(seconds=self.stalled_grace_seconds)
|
||||||
|
stalled_txs = get_status_tx(
|
||||||
|
StatusEnum.SENT.value,
|
||||||
|
before=before,
|
||||||
|
)
|
||||||
|
# return list(failed_txs.keys()) + list(stalled_txs.keys())
|
||||||
|
return stalled_txs
|
||||||
|
|
||||||
|
|
||||||
|
def process(self, w3, ref):
|
||||||
|
logg.debug('tx {}'.format(ref))
|
||||||
|
for f in self.filter:
|
||||||
|
f(w3, ref, None, str(self.chain()))
|
||||||
|
|
||||||
|
|
||||||
|
def loop(self, interval):
|
||||||
|
chain_str = str(self.chain())
|
||||||
|
while self.running and Syncer.running_global:
|
||||||
|
c = RpcClient(self.chain())
|
||||||
|
for tx in self.get(c.w3):
|
||||||
|
self.process(c.w3, tx)
|
||||||
|
if self.final_func != None:
|
||||||
|
self.final_func(chain_str)
|
||||||
|
time.sleep(interval)
|
@ -20,4 +20,5 @@ moolb~=0.1.1b2
|
|||||||
eth-address-index~=0.1.0a8
|
eth-address-index~=0.1.0a8
|
||||||
chainlib~=0.0.1a17
|
chainlib~=0.0.1a17
|
||||||
hexathon~=0.0.1a3
|
hexathon~=0.0.1a3
|
||||||
chainsyncer~=0.0.1a12
|
chainsyncer~=0.0.1a15
|
||||||
|
cic-base==0.1.1a3
|
||||||
|
@ -105,63 +105,14 @@ RUN cd root && \
|
|||||||
RUN cd cic-bancor/python && \
|
RUN cd cic-bancor/python && \
|
||||||
pip install --extra-index-url $pip_extra_index_url .
|
pip install --extra-index-url $pip_extra_index_url .
|
||||||
|
|
||||||
RUN echo installing common python tooling
|
|
||||||
ARG cic_python_commit=9bd1d5319aad8791ebf353707eabbb7b3e7ab5d0
|
|
||||||
ARG cic_python_url=https://gitlab.com/grassrootseconomics/cic-python.git/
|
|
||||||
RUN echo Install sum of python dependencies across all components && \
|
|
||||||
git clone --depth 1 $cic_python_url cic-python && \
|
|
||||||
cd cic-python && \
|
|
||||||
git fetch --depth 1 origin $cic_python_commit && \
|
|
||||||
git checkout $cic_python_commit && \
|
|
||||||
pip install --extra-index-url $pip_extra_index_url -r requirements.txt
|
|
||||||
|
|
||||||
RUN echo Install dev-only provisions
|
RUN apt-get install -y cargo
|
||||||
ARG cryptocurrency_cli_tools_version=0.0.4
|
ARG cic_base_version=0.1.1a3
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url cryptocurrency-cli-tools==$cryptocurrency_cli_tools_version
|
RUN pip install --extra-index-url $pip_extra_index_url cic-base[full_graph]==$cic_base_version
|
||||||
|
|
||||||
RUN echo Install smart contract interface implementations, least frequently changed first
|
ARG cic_registry_version=0.5.3a21
|
||||||
ARG giftable_erc20_token_version=0.0.7b12
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url giftable-erc20-token==$giftable_erc20_token_version
|
|
||||||
|
|
||||||
ARG eth_accounts_index_version=0.0.10a9
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url eth-accounts-index==$eth_accounts_index_version
|
|
||||||
|
|
||||||
#ARG erc20_approval_escrow_version=0.3.0a4
|
|
||||||
#RUN pip install --extra-index-url $pip_extra_index_url erc20-approval-escrow==$erc20_approval_escrow_version
|
|
||||||
ARG erc20_transfer_authorization_version=0.3.0a9
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url erc20-transfer-authorization==$erc20_transfer_authorization_version
|
|
||||||
|
|
||||||
#ARG erc20_single_shot_faucet_version=0.2.0a5
|
|
||||||
#RUN pip install --extra-index-url $pip_extra_index_url erc20-single-shot-faucet==$erc20_single_shot_faucet_version
|
|
||||||
|
|
||||||
ARG sarafu_faucet_version=0.0.1a11
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url sarafu-faucet==$sarafu_faucet_version
|
|
||||||
|
|
||||||
ARG eth_address_index_version=0.1.0a10
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url eth-address-index==$eth_address_index_version
|
|
||||||
|
|
||||||
RUN echo Install cic specific python packages
|
|
||||||
ARG cic_registry_version=0.5.3a20+build.30d0026c
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url cic-registry==$cic_registry_version
|
RUN pip install --extra-index-url $pip_extra_index_url cic-registry==$cic_registry_version
|
||||||
|
|
||||||
RUN echo Install misc helpers
|
|
||||||
|
|
||||||
ARG crypto_dev_signer_version=0.4.13rc2
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url crypto-dev-signer==$crypto_dev_signer_version
|
|
||||||
|
|
||||||
ARG eth_gas_proxy_version=0.0.1a4
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url eth-gas-proxy==$eth_gas_proxy_version
|
|
||||||
|
|
||||||
ARG cic_contracts_version=0.0.2a2
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url cic-contracts==$cic_contracts_version
|
|
||||||
|
|
||||||
ARG chainlib_version=0.0.1a17
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url chainlib==$chainlib_version
|
|
||||||
|
|
||||||
ARG chainsyncer_version=0.0.1a12
|
|
||||||
RUN pip install --extra-index-url $pip_extra_index_url chainsyncer==$chainsyncer_version
|
|
||||||
|
|
||||||
|
|
||||||
WORKDIR /root
|
WORKDIR /root
|
||||||
|
|
||||||
COPY contract-migration/testdata/pgp testdata/pgp
|
COPY contract-migration/testdata/pgp testdata/pgp
|
||||||
@ -177,6 +128,4 @@ COPY contract-migration/seed_cic_eth.sh seed_cic_eth.sh
|
|||||||
COPY contract-migration/sarafu_declaration.json sarafu_declaration.json
|
COPY contract-migration/sarafu_declaration.json sarafu_declaration.json
|
||||||
COPY contract-migration/keystore keystore
|
COPY contract-migration/keystore keystore
|
||||||
|
|
||||||
LABEL version="4"
|
|
||||||
|
|
||||||
ENTRYPOINT [ "/bin/bash" ]
|
ENTRYPOINT [ "/bin/bash" ]
|
||||||
|
@ -1,12 +1 @@
|
|||||||
psycopg2==2.8.6
|
cic-base[full]==0.1.1a1
|
||||||
chainlib~=0.0.1a17
|
|
||||||
chainsyncer~=0.0.1a12
|
|
||||||
cic-eth==0.10.0a31
|
|
||||||
cic-registry~=0.5.3a21
|
|
||||||
confini~=0.3.6rc3
|
|
||||||
celery==4.4.7
|
|
||||||
redis==3.5.3
|
|
||||||
hexathon~=0.0.1a3
|
|
||||||
faker==4.17.1
|
|
||||||
cic-types==0.1.0a7+build.1c254367
|
|
||||||
eth-accounts-index~=0.0.10a10
|
|
||||||
|
@ -43,5 +43,5 @@ cryptocurrency-cli-tools==0.0.4
|
|||||||
giftable-erc20-token==0.0.7b12
|
giftable-erc20-token==0.0.7b12
|
||||||
hexathon==0.0.1a3
|
hexathon==0.0.1a3
|
||||||
chainlib==0.0.1a17
|
chainlib==0.0.1a17
|
||||||
chainsyncer==0.0.1a12
|
chainsyncer==0.0.1a15
|
||||||
cic-registry==0.5.3.a21
|
cic-registry==0.5.3.a21
|
||||||
|
@ -240,7 +240,7 @@ services:
|
|||||||
./start_tasker.sh -q cic-eth -vv
|
./start_tasker.sh -q cic-eth -vv
|
||||||
# command: [/bin/sh, "./start_tasker.sh", -q, cic-eth, -vv ]
|
# command: [/bin/sh, "./start_tasker.sh", -q, cic-eth, -vv ]
|
||||||
|
|
||||||
cic-eth-manager-head:
|
cic-eth-tracker:
|
||||||
build:
|
build:
|
||||||
context: apps/
|
context: apps/
|
||||||
dockerfile: cic-eth/docker/Dockerfile
|
dockerfile: cic-eth/docker/Dockerfile
|
||||||
@ -274,45 +274,9 @@ services:
|
|||||||
- -c
|
- -c
|
||||||
- |
|
- |
|
||||||
if [[ -f /tmp/cic/config/.env ]]; then source /tmp/cic/config/.env; fi
|
if [[ -f /tmp/cic/config/.env ]]; then source /tmp/cic/config/.env; fi
|
||||||
./start_manager.sh head -v
|
./start_manager.sh -v -c /usr/local/etc/cic-eth
|
||||||
# command: "/root/start_manager.sh head -vv"
|
# command: "/root/start_manager.sh head -vv"
|
||||||
|
|
||||||
cic-eth-manager-history:
|
|
||||||
build:
|
|
||||||
context: apps/
|
|
||||||
dockerfile: cic-eth/docker/Dockerfile
|
|
||||||
environment:
|
|
||||||
ETH_PROVIDER: http://eth:8545
|
|
||||||
DATABASE_USER: ${DATABASE_USER:-grassroots}
|
|
||||||
DATABASE_HOST: ${DATABASE_HOST:-postgres}
|
|
||||||
DATABASE_PASSWORD: ${DATABASE_PASSWORD:-tralala}
|
|
||||||
DATABASE_NAME: ${DATABASE_NAME_CIC_CACHE:-cic_eth}
|
|
||||||
DATABASE_PORT: ${DATABASE_PORT:-5432}
|
|
||||||
DATABASE_ENGINE: ${DATABASE_ENGINE:-postgres}
|
|
||||||
DATABASE_DRIVER: ${DATABASE_DRIVER:-psycopg2}
|
|
||||||
DATABASE_DEBUG: ${DATABASE_DEBUG:-0}
|
|
||||||
CIC_CHAIN_SPEC: ${CIC_CHAIN_SPEC:-evm:bloxberg:8996}
|
|
||||||
CIC_REGISTRY_ADDRESS: $CIC_REGISTRY_ADDRESS
|
|
||||||
#BANCOR_DIR: $BANCOR_DIR
|
|
||||||
CELERY_BROKER_URL: ${CELERY_BROKER_URL:-redis://redis}
|
|
||||||
CELERY_RESULT_URL: ${CELERY_RESULT_URL:-redis://redis}
|
|
||||||
TASKS_TRANSFER_CALLBACKS: $TASKS_TRANSFER_CALLBACKS
|
|
||||||
depends_on:
|
|
||||||
- eth
|
|
||||||
- postgres
|
|
||||||
- redis
|
|
||||||
#deploy:
|
|
||||||
#restart_policy:
|
|
||||||
# condition: on-failure
|
|
||||||
volumes:
|
|
||||||
- contract-config:/tmp/cic/config/:ro
|
|
||||||
command:
|
|
||||||
- /bin/bash
|
|
||||||
- -c
|
|
||||||
- |
|
|
||||||
if [[ -f /tmp/cic/config/.env ]]; then source /tmp/cic/config/.env; fi
|
|
||||||
./start_manager.sh history -v
|
|
||||||
# command: "/root/start_manager.sh history -vv"
|
|
||||||
|
|
||||||
cic-eth-dispatcher:
|
cic-eth-dispatcher:
|
||||||
build:
|
build:
|
||||||
|
Loading…
Reference in New Issue
Block a user