import os import asyncio import logging from binascii import unhexlify from typing import Optional, Tuple, Set, List, Coroutine from concurrent.futures import ThreadPoolExecutor from lbry.db import Database, trending from lbry.db import queries as q from lbry.db.constants import TXO_TYPES, CLAIM_TYPE_CODES from lbry.db.query_context import Event, Progress from lbry.event import BroadcastSubscription, EventController from lbry.service.base import Sync, BlockEvent from lbry.blockchain.lbrycrd import Lbrycrd from lbry.error import LbrycrdEventSubscriptionError from . import blocks as block_phase, claims as claim_phase, supports as support_phase from .context import uninitialize from .filter_builder import split_range_into_10k_batches log = logging.getLogger(__name__) BLOCKS_INIT_EVENT = Event.add("blockchain.sync.blocks.init", "steps") BLOCKS_MAIN_EVENT = Event.add("blockchain.sync.blocks.main", "blocks", "txs") FILTER_INIT_EVENT = Event.add("blockchain.sync.filters.init", "steps") FILTER_MAIN_EVENT = Event.add("blockchain.sync.filters.main", "blocks") CLAIMS_INIT_EVENT = Event.add("blockchain.sync.claims.init", "steps") CLAIMS_MAIN_EVENT = Event.add("blockchain.sync.claims.main", "claims") TRENDS_INIT_EVENT = Event.add("blockchain.sync.trends.init", "steps") TRENDS_MAIN_EVENT = Event.add("blockchain.sync.trends.main", "blocks") SUPPORTS_INIT_EVENT = Event.add("blockchain.sync.supports.init", "steps") SUPPORTS_MAIN_EVENT = Event.add("blockchain.sync.supports.main", "supports") class BlockchainSync(Sync): TX_FLUSH_SIZE = 25_000 # flush to db after processing this many TXs and update progress CLAIM_FLUSH_SIZE = 25_000 # flush to db after processing this many claims and update progress SUPPORT_FLUSH_SIZE = 25_000 # flush to db after processing this many supports and update progress FILTER_FLUSH_SIZE = 10_000 # flush to db after processing this many filters and update progress def __init__(self, chain: Lbrycrd, db: Database): super().__init__(chain.ledger, db) self.chain = chain self.pid = os.getpid() self._on_block_controller = EventController() self.on_block = self._on_block_controller.stream self.conf.events.register("blockchain.block", self.on_block) self._on_mempool_controller = EventController() self.on_mempool = self._on_mempool_controller.stream self.on_block_hash_subscription: Optional[BroadcastSubscription] = None self.on_tx_hash_subscription: Optional[BroadcastSubscription] = None self.advance_loop_task: Optional[asyncio.Task] = None self.block_hash_event = asyncio.Event() self.tx_hash_event = asyncio.Event() self.mempool = [] self.search_censor_channel_hashes = { unhexlify(channel_id)[::-1] for channel_id in self.conf.search_censor_channel_ids } self.resolve_censor_channel_hashes = { unhexlify(channel_id)[::-1] for channel_id in self.conf.resolve_censor_channel_ids } async def wait_for_chain_ready(self): while True: try: return await self.chain.ensure_subscribable() except asyncio.CancelledError: raise except LbrycrdEventSubscriptionError as e: log.warning( "Lbrycrd is misconfigured. Please double check if" " zmqpubhashblock is properly set on lbrycrd.conf" ) raise except Exception as e: log.warning("Blockchain not ready, waiting for it: %s", str(e)) await asyncio.sleep(1) async def start(self): self.db.stop_event.clear() await self.wait_for_chain_ready() self.advance_loop_task = asyncio.create_task(self.advance()) await self.advance_loop_task await self.chain.subscribe() self.advance_loop_task = asyncio.create_task(self.advance_loop()) self.on_block_hash_subscription = self.chain.on_block_hash.listen( lambda e: self.block_hash_event.set() ) self.on_tx_hash_subscription = self.chain.on_tx_hash.listen( lambda e: self.tx_hash_event.set() ) async def stop(self): self.chain.unsubscribe() self.db.stop_event.set() for subscription in ( self.on_block_hash_subscription, self.on_tx_hash_subscription, self.advance_loop_task ): if subscription is not None: subscription.cancel() if isinstance(self.db.executor, ThreadPoolExecutor): await self.db.run(uninitialize) async def run_tasks(self, tasks: List[Coroutine]) -> Optional[Set[asyncio.Future]]: done, pending = await asyncio.wait( tasks, return_when=asyncio.FIRST_EXCEPTION ) if pending: self.db.stop_event.set() for future in pending: future.cancel() for future in done: future.result() return return done async def get_block_headers(self, start_height: int, end_height: int = None): return await self.db.get_block_headers(start_height, end_height) async def get_best_block_height(self) -> int: return await self.db.get_best_block_height() async def get_best_block_height_for_file(self, file_number) -> int: return await self.db.run( block_phase.get_best_block_height_for_file, file_number ) async def sync_blocks(self) -> Optional[Tuple[int, int]]: tasks = [] starting_height = None tx_count = block_count = 0 with Progress(self.db.message_queue, BLOCKS_INIT_EVENT) as p: ending_height = await self.chain.db.get_best_height() for chain_file in p.iter(await self.chain.db.get_block_files()): # block files may be read and saved out of order, need to check # each file individually to see if we have missing blocks our_best_file_height = await self.get_best_block_height_for_file( chain_file['file_number'] ) if our_best_file_height == chain_file['best_height']: # we have all blocks in this file, skipping continue if -1 < our_best_file_height < chain_file['best_height']: # we have some blocks, need to figure out what we're missing # call get_block_files again limited to this file and current_height chain_file = (await self.chain.db.get_block_files( file_number=chain_file['file_number'], start_height=our_best_file_height+1, ))[0] tx_count += chain_file['txs'] block_count += chain_file['blocks'] file_start_height = chain_file['start_height'] starting_height = min( file_start_height if starting_height is None else starting_height, file_start_height ) tasks.append(self.db.run( block_phase.sync_block_file, chain_file['file_number'], file_start_height, chain_file['txs'], self.TX_FLUSH_SIZE )) with Progress(self.db.message_queue, BLOCKS_MAIN_EVENT) as p: p.start(block_count, tx_count, extra={ "starting_height": starting_height, "ending_height": ending_height, "files": len(tasks), "claims": await self.chain.db.get_claim_metadata_count(starting_height, ending_height), "supports": await self.chain.db.get_support_metadata_count(starting_height, ending_height), }) completed = await self.run_tasks(tasks) if completed: if starting_height == 0: await self.db.run(block_phase.blocks_constraints_and_indexes) else: await self.db.run(block_phase.blocks_vacuum) best_height_processed = max(f.result() for f in completed) return starting_height, best_height_processed async def sync_filters(self): with Progress(self.db.message_queue, FILTER_INIT_EVENT) as p: p.start(2) initial_sync = not await self.db.has_filters() p.step() if initial_sync: blocks = [0, await self.db.get_best_block_height()] else: blocks = await self.db.run(block_phase.get_block_range_without_filters) if blocks != (-1, -1): batches = split_range_into_10k_batches(*blocks) p.step() else: p.step() return with Progress(self.db.message_queue, FILTER_MAIN_EVENT) as p: p.start((blocks[1]-blocks[0])+1) await self.run_tasks([ self.db.run(block_phase.sync_filters, *batch) for batch in batches ]) if initial_sync: await self.db.run(block_phase.filters_constraints_and_indexes) else: await self.db.run(block_phase.filters_vacuum) async def sync_spends(self, blocks_added): if blocks_added: await self.db.run(block_phase.sync_spends, blocks_added[0] == 0) async def count_unspent_txos( self, txo_types: Tuple[int, ...], blocks: Tuple[int, int] = None, missing_in_supports_table: bool = False, missing_in_claims_table: bool = False, missing_or_stale_in_claims_table: bool = False, ) -> int: return await self.db.run( q.count_unspent_txos, txo_types, blocks, missing_in_supports_table, missing_in_claims_table, missing_or_stale_in_claims_table, ) async def distribute_unspent_txos( self, txo_types: Tuple[int, ...], blocks: Tuple[int, int] = None, missing_in_supports_table: bool = False, missing_in_claims_table: bool = False, missing_or_stale_in_claims_table: bool = False, ) -> int: return await self.db.run( q.distribute_unspent_txos, txo_types, blocks, missing_in_supports_table, missing_in_claims_table, missing_or_stale_in_claims_table, self.db.workers ) async def count_abandoned_supports(self) -> int: return await self.db.run(q.count_abandoned_supports) async def count_abandoned_claims(self) -> int: return await self.db.run(q.count_abandoned_claims) async def count_claims_with_changed_supports(self, blocks) -> int: return await self.db.run(q.count_claims_with_changed_supports, blocks) async def count_claims_with_changed_reposts(self, blocks) -> int: return await self.db.run(q.count_claims_with_changed_reposts, blocks) async def count_claims_with_stale_signatures(self, blocks) -> int: return await self.db.run(q.count_claims_with_stale_signatures, blocks) async def count_channels_with_changed_content(self, blocks) -> int: return await self.db.run(q.count_channels_with_changed_content, blocks) async def count_takeovers(self, blocks) -> int: return await self.chain.db.get_takeover_count( start_height=blocks[0], end_height=blocks[-1] ) async def sync_claims(self, blocks) -> bool: delete_claims = takeovers = claims_with_changed_supports =\ claims_with_changed_reposts = claims_with_stale_signatures = 0 initial_sync = not await self.db.has_claims() with Progress(self.db.message_queue, CLAIMS_INIT_EVENT) as p: if initial_sync: total, batches = await self.distribute_unspent_txos(CLAIM_TYPE_CODES) elif blocks: p.start(6) # 1. content claims to be inserted or updated total = await self.count_unspent_txos( CLAIM_TYPE_CODES, blocks, missing_or_stale_in_claims_table=True ) batches = [blocks] if total else [] p.step() # 2. claims to be deleted delete_claims = await self.count_abandoned_claims() total += delete_claims p.step() # 3. claims to be updated with new support totals claims_with_changed_supports = await self.count_claims_with_changed_supports(blocks) total += claims_with_changed_supports p.step() # 4. claims to be updated with new repost totals claims_with_changed_reposts = await self.count_claims_with_changed_reposts(blocks) total += claims_with_changed_reposts p.step() # 5. claims to be updated due to name takeovers takeovers = await self.count_takeovers(blocks) total += takeovers p.step() # 6. claims where channel signature changed and claim was not re-signed in time claims_with_stale_signatures = await self.count_claims_with_stale_signatures(blocks) total += claims_with_stale_signatures p.step() else: return initial_sync with Progress(self.db.message_queue, CLAIMS_MAIN_EVENT) as p: p.start(total) if batches: await self.run_tasks([ self.db.run(claim_phase.claims_insert, batch, not initial_sync, self.CLAIM_FLUSH_SIZE) for batch in batches ]) if not initial_sync: await self.run_tasks([ self.db.run(claim_phase.claims_update, batch) for batch in batches ]) if delete_claims: await self.db.run(claim_phase.claims_delete, delete_claims) if takeovers: await self.db.run(claim_phase.update_takeovers, blocks, takeovers) if claims_with_changed_supports: await self.db.run(claim_phase.update_stakes, blocks, claims_with_changed_supports) if claims_with_changed_reposts: await self.db.run(claim_phase.update_reposts, blocks, claims_with_changed_reposts) if claims_with_stale_signatures: await self.db.run(claim_phase.update_stale_signatures, blocks, claims_with_stale_signatures) if initial_sync: await self.db.run(claim_phase.claims_constraints_and_indexes) else: await self.db.run(claim_phase.claims_vacuum) return initial_sync async def sync_supports(self, blocks): delete_supports = 0 initial_sync = not await self.db.has_supports() with Progress(self.db.message_queue, SUPPORTS_INIT_EVENT) as p: if initial_sync: total, support_batches = await self.distribute_unspent_txos(TXO_TYPES['support']) elif blocks: p.start(2) # 1. supports to be inserted total = await self.count_unspent_txos( TXO_TYPES['support'], blocks, missing_in_supports_table=True ) support_batches = [blocks] if total else [] p.step() # 2. supports to be deleted delete_supports = await self.count_abandoned_supports() total += delete_supports p.step() else: return with Progress(self.db.message_queue, SUPPORTS_MAIN_EVENT) as p: p.start(total) if support_batches: await self.run_tasks([ self.db.run( support_phase.supports_insert, batch, not initial_sync, self.SUPPORT_FLUSH_SIZE ) for batch in support_batches ]) if delete_supports: await self.db.run(support_phase.supports_delete, delete_supports) if initial_sync: await self.db.run(support_phase.supports_constraints_and_indexes) else: await self.db.run(support_phase.supports_vacuum) async def sync_channel_stats(self, blocks, initial_sync): await self.db.run(claim_phase.update_channel_stats, blocks, initial_sync) async def sync_trends(self): ending_height = await self.chain.db.get_best_height() if ending_height is not None: await self.db.run(trending.calculate_trending, ending_height) async def sync_claim_filtering(self): await self.db.run( claim_phase.update_claim_filters, self.resolve_censor_channel_hashes, self.search_censor_channel_hashes ) async def advance(self): blocks_added = await self.sync_blocks() await self.sync_spends(blocks_added) await self.sync_filters() initial_claim_sync = await self.sync_claims(blocks_added) await self.sync_supports(blocks_added) await self.sync_channel_stats(blocks_added, initial_claim_sync) await self.sync_trends() await self.sync_claim_filtering() if blocks_added: await self._on_block_controller.add(BlockEvent(blocks_added[-1])) async def sync_mempool(self): added = await self.db.run(block_phase.sync_mempool) await self.sync_spends([-1]) await self.db.run(claim_phase.claims_insert, [-1, -1], True, self.CLAIM_FLUSH_SIZE) await self.db.run(claim_phase.claims_update, [-1, -1]) await self.db.run(claim_phase.claims_vacuum) self.mempool.extend(added) await self._on_mempool_controller.add(added) async def clear_mempool(self): self.mempool.clear() await self.db.run(block_phase.clear_mempool) async def advance_loop(self): while True: try: await asyncio.wait([ self.tx_hash_event.wait(), self.block_hash_event.wait(), ], return_when=asyncio.FIRST_COMPLETED) if self.block_hash_event.is_set(): self.block_hash_event.clear() #await self.clear_mempool() await self.advance() self.tx_hash_event.clear() #await self.sync_mempool() except asyncio.CancelledError: return except Exception as e: log.exception(e) await self.stop() async def rewind(self, height): await self.db.run(block_phase.rewind, height)