262 lines
10 KiB
Python
262 lines
10 KiB
Python
import logging
|
|
import asyncio
|
|
import typing
|
|
import signal
|
|
from concurrent.futures.thread import ThreadPoolExecutor
|
|
from prometheus_client import Gauge, Histogram
|
|
|
|
from scribe import __version__, PROMETHEUS_NAMESPACE
|
|
from scribe.env import Env
|
|
from scribe.db import HubDB
|
|
from scribe.db.prefixes import DBState
|
|
from scribe.common import HISTOGRAM_BUCKETS
|
|
from scribe.metrics import PrometheusServer
|
|
|
|
|
|
class BlockchainService:
|
|
"""
|
|
Base class for blockchain readers as well as the block processor
|
|
"""
|
|
def __init__(self, env: Env, secondary_name: str, thread_workers: int = 1, thread_prefix: str = 'scribe'):
|
|
self.env = env
|
|
self.log = logging.getLogger(__name__).getChild(self.__class__.__name__)
|
|
self.shutdown_event = asyncio.Event()
|
|
self.cancellable_tasks = []
|
|
self._thread_workers = thread_workers
|
|
self._thread_prefix = thread_prefix
|
|
self._executor = ThreadPoolExecutor(thread_workers, thread_name_prefix=thread_prefix)
|
|
self.lock = asyncio.Lock()
|
|
self.last_state: typing.Optional[DBState] = None
|
|
self.db = HubDB(
|
|
env.coin, env.db_dir, env.reorg_limit, env.cache_all_claim_txos, env.cache_all_tx_hashes,
|
|
secondary_name=secondary_name, max_open_files=-1, blocking_channel_ids=env.blocking_channel_ids,
|
|
filtering_channel_ids=env.filtering_channel_ids, executor=self._executor
|
|
)
|
|
self._stopping = False
|
|
|
|
def start_cancellable(self, run, *args):
|
|
_flag = asyncio.Event()
|
|
self.cancellable_tasks.append(asyncio.ensure_future(run(*args, _flag)))
|
|
return _flag.wait()
|
|
|
|
def _iter_start_tasks(self):
|
|
raise NotImplementedError()
|
|
|
|
def _iter_stop_tasks(self):
|
|
yield self._stop_cancellable_tasks()
|
|
|
|
async def _stop_cancellable_tasks(self):
|
|
async with self.lock:
|
|
while self.cancellable_tasks:
|
|
t = self.cancellable_tasks.pop()
|
|
if not t.done():
|
|
t.cancel()
|
|
|
|
async def start(self):
|
|
if not self._executor:
|
|
self._executor = ThreadPoolExecutor(self._thread_workers, thread_name_prefix=self._thread_prefix)
|
|
self.db._executor = self._executor
|
|
|
|
env = self.env
|
|
# min_str, max_str = env.coin.SESSIONCLS.protocol_min_max_strings()
|
|
self.log.info(f'software version: {__version__}')
|
|
# self.log.info(f'supported protocol versions: {min_str}-{max_str}')
|
|
self.log.info(f'reorg limit is {env.reorg_limit:,d} blocks')
|
|
|
|
self.db.open_db()
|
|
self.log.info(f'initializing caches')
|
|
await self.db.initialize_caches()
|
|
self.last_state = self.db.read_db_state()
|
|
self.log.info(f'opened db at block {self.db.db_height}')
|
|
|
|
for start_task in self._iter_start_tasks():
|
|
await start_task
|
|
|
|
async def stop(self):
|
|
self.log.info("stopping")
|
|
self._stopping = True
|
|
for stop_task in self._iter_stop_tasks():
|
|
await stop_task
|
|
self.db.close()
|
|
self._executor.shutdown(wait=True)
|
|
self._executor = None
|
|
self.shutdown_event.set()
|
|
|
|
async def _run(self):
|
|
try:
|
|
await self.start()
|
|
await self.shutdown_event.wait()
|
|
except (SystemExit, KeyboardInterrupt, asyncio.CancelledError):
|
|
self._stopping = True
|
|
except Exception as err:
|
|
self.log.exception("unexpected fatal error: %s", err)
|
|
self._stopping = True
|
|
|
|
def run(self):
|
|
def __exit():
|
|
raise SystemExit()
|
|
|
|
loop = asyncio.get_event_loop()
|
|
loop.set_default_executor(self._executor)
|
|
loop.add_signal_handler(signal.SIGINT, __exit)
|
|
loop.add_signal_handler(signal.SIGTERM, __exit)
|
|
try:
|
|
loop.run_until_complete(self._run())
|
|
finally:
|
|
loop.run_until_complete(self.stop())
|
|
|
|
|
|
NAMESPACE = f"{PROMETHEUS_NAMESPACE}_reader"
|
|
|
|
|
|
class BlockchainReaderService(BlockchainService):
|
|
block_count_metric = Gauge(
|
|
"block_count", "Number of processed blocks", namespace=NAMESPACE
|
|
)
|
|
block_update_time_metric = Histogram(
|
|
"block_time", "Block update times", namespace=NAMESPACE, buckets=HISTOGRAM_BUCKETS
|
|
)
|
|
reorg_count_metric = Gauge(
|
|
"reorg_count", "Number of reorgs", namespace=NAMESPACE
|
|
)
|
|
|
|
def __init__(self, env, secondary_name: str, thread_workers: int = 1, thread_prefix: str = 'blockchain-reader'):
|
|
super().__init__(env, secondary_name, thread_workers, thread_prefix)
|
|
self._refresh_interval = 0.1
|
|
self.prometheus_server: typing.Optional[PrometheusServer] = None
|
|
self.finished_initial_catch_up = asyncio.Event()
|
|
|
|
async def poll_for_changes(self):
|
|
"""
|
|
Detect and handle if the db has advanced to a new block or unwound during a chain reorganization
|
|
|
|
If a reorg is detected, this will first unwind() to the branching height and then advance() forward
|
|
to the new block(s).
|
|
"""
|
|
await asyncio.get_event_loop().run_in_executor(self._executor, self._detect_changes)
|
|
|
|
def clear_caches(self):
|
|
"""
|
|
Called after finished advancing, used for invalidating caches
|
|
"""
|
|
pass
|
|
|
|
def advance(self, height: int):
|
|
"""
|
|
Called when advancing to the given block height
|
|
Callbacks that look up new values from the added block can be put here
|
|
"""
|
|
tx_count = self.db.prefix_db.tx_count.get(height).tx_count
|
|
assert tx_count not in self.db.tx_counts, f'boom {tx_count} in {len(self.db.tx_counts)} tx counts'
|
|
assert len(self.db.tx_counts) == height, f"{len(self.db.tx_counts)} != {height}"
|
|
prev_count = self.db.tx_counts[-1]
|
|
self.db.tx_counts.append(tx_count)
|
|
|
|
# precache all of the txs from this block
|
|
block_tx_hashes = self.db.prefix_db.block_txs.get(height).tx_hashes
|
|
block_txs = self.db.prefix_db.tx.multi_get(
|
|
[(tx_hash,) for tx_hash in block_tx_hashes], deserialize_value=False
|
|
)
|
|
for tx_pos, (tx_num, tx_hash, tx) in enumerate(zip(range(prev_count, tx_count), block_tx_hashes, block_txs)):
|
|
self.db.tx_cache[tx_hash] = tx, tx_num, tx_pos, height
|
|
if self.db._cache_all_tx_hashes:
|
|
self.db.total_transactions.append(tx_hash)
|
|
self.db.tx_num_mapping[tx_hash] = tx_count
|
|
if self.db._cache_all_tx_hashes:
|
|
assert len(self.db.total_transactions) == tx_count, f"{len(self.db.total_transactions)} vs {tx_count}"
|
|
|
|
header = self.db.prefix_db.header.get(height, deserialize_value=False)
|
|
self.db.headers.append(header)
|
|
self.db.block_hashes.append(self.env.coin.header_hash(header))
|
|
|
|
def unwind(self):
|
|
"""
|
|
Go backwards one block
|
|
"""
|
|
prev_count = self.db.tx_counts.pop()
|
|
tx_count = self.db.tx_counts[-1]
|
|
self.db.headers.pop()
|
|
self.db.block_hashes.pop()
|
|
if self.db._cache_all_tx_hashes:
|
|
for _ in range(prev_count - tx_count):
|
|
tx_hash = self.db.tx_num_mapping.pop(self.db.total_transactions.pop())
|
|
if tx_hash in self.db.tx_cache:
|
|
self.db.tx_cache.pop(tx_hash)
|
|
assert len(self.db.total_transactions) == tx_count, f"{len(self.db.total_transactions)} vs {tx_count}"
|
|
self.db.merkle_cache.clear()
|
|
|
|
def _detect_changes(self):
|
|
try:
|
|
self.db.prefix_db.try_catch_up_with_primary()
|
|
except Exception as err:
|
|
self.log.exception('failed to update secondary db: %s', err)
|
|
raise err
|
|
state = self.db.prefix_db.db_state.get()
|
|
if not state or state.height <= 0:
|
|
return
|
|
if self.last_state and self.last_state.height > state.height:
|
|
self.log.warning("reorg detected, waiting until the writer has flushed the new blocks to advance")
|
|
return
|
|
last_height = 0 if not self.last_state else self.last_state.height
|
|
rewound = False
|
|
if self.last_state:
|
|
while True:
|
|
if self.db.headers[-1] == self.db.prefix_db.header.get(last_height, deserialize_value=False):
|
|
self.log.debug("connects to block %i", last_height)
|
|
break
|
|
else:
|
|
self.log.warning("disconnect block %i", last_height)
|
|
self.unwind()
|
|
rewound = True
|
|
last_height -= 1
|
|
if rewound:
|
|
self.reorg_count_metric.inc()
|
|
self.db.read_db_state()
|
|
if not self.last_state or last_height < state.height:
|
|
for height in range(last_height + 1, state.height + 1):
|
|
self.log.info("advancing to %i", height)
|
|
self.advance(height)
|
|
self.clear_caches()
|
|
self.last_state = state
|
|
self.block_count_metric.set(self.last_state.height)
|
|
self.db.blocked_streams, self.db.blocked_channels = self.db.get_streams_and_channels_reposted_by_channel_hashes(
|
|
self.db.blocking_channel_hashes
|
|
)
|
|
self.db.filtered_streams, self.db.filtered_channels = self.db.get_streams_and_channels_reposted_by_channel_hashes(
|
|
self.db.filtering_channel_hashes
|
|
)
|
|
|
|
async def refresh_blocks_forever(self, synchronized: asyncio.Event):
|
|
while True:
|
|
try:
|
|
async with self.lock:
|
|
await self.poll_for_changes()
|
|
if not self.db.catching_up and not self.finished_initial_catch_up.is_set():
|
|
self.finished_initial_catch_up.set()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except:
|
|
self.log.exception("blockchain reader main loop encountered an unexpected error")
|
|
self.shutdown_event.set()
|
|
raise
|
|
await asyncio.sleep(self._refresh_interval)
|
|
synchronized.set()
|
|
|
|
def _iter_start_tasks(self):
|
|
self.block_count_metric.set(self.last_state.height)
|
|
yield self.start_prometheus()
|
|
yield self.start_cancellable(self.refresh_blocks_forever)
|
|
|
|
def _iter_stop_tasks(self):
|
|
yield self.stop_prometheus()
|
|
yield self._stop_cancellable_tasks()
|
|
|
|
async def start_prometheus(self):
|
|
if not self.prometheus_server and self.env.prometheus_port:
|
|
self.prometheus_server = PrometheusServer()
|
|
await self.prometheus_server.start("0.0.0.0", self.env.prometheus_port)
|
|
|
|
async def stop_prometheus(self):
|
|
if self.prometheus_server:
|
|
await self.prometheus_server.stop()
|
|
self.prometheus_server = None
|