From 7a4e5dcb0598f1319274374af42922031d1af8de Mon Sep 17 00:00:00 2001 From: Lex Berezhny Date: Sun, 12 Jul 2020 12:02:58 -0400 Subject: [PATCH] handle claims in abandoned channels --- lbry/blockchain/sync/blocks.py | 2 +- lbry/blockchain/sync/synchronizer.py | 3 ++- lbry/console.py | 16 +++++++++++----- lbry/db/query_context.py | 2 +- tests/integration/blockchain/test_blockchain.py | 17 +++++++++++++++++ 5 files changed, 32 insertions(+), 8 deletions(-) diff --git a/lbry/blockchain/sync/blocks.py b/lbry/blockchain/sync/blocks.py index e2be98184..778979cac 100644 --- a/lbry/blockchain/sync/blocks.py +++ b/lbry/blockchain/sync/blocks.py @@ -27,7 +27,7 @@ def get_best_block_height_for_file(file_number): )['height'] -@event_emitter("blockchain.sync.block.file", "blocks", "txs", throttle=50) +@event_emitter("blockchain.sync.block.file", "blocks", "txs", throttle=100) def sync_block_file( file_number: int, start_height: int, txs: int, flush_size: int, p: ProgressContext ): diff --git a/lbry/blockchain/sync/synchronizer.py b/lbry/blockchain/sync/synchronizer.py index 9c5073f18..f641d1fac 100644 --- a/lbry/blockchain/sync/synchronizer.py +++ b/lbry/blockchain/sync/synchronizer.py @@ -31,7 +31,7 @@ TREND_MAIN_EVENT = Event.add("blockchain.sync.trends.main", "blocks") class BlockchainSync(Sync): - TX_FLUSH_SIZE = 20_000 # flush to db after processing this many TXs and update progress + TX_FLUSH_SIZE = 100_000 # flush to db after processing this many TXs and update progress FILTER_CHUNK_SIZE = 100_000 # split filter generation tasks into this size block chunks FILTER_FLUSH_SIZE = 10_000 # flush to db after processing this many filters and update progress CLAIM_CHUNK_SIZE = 50_000 # split claim sync tasks into this size block chunks @@ -48,6 +48,7 @@ class BlockchainSync(Sync): self.advance_loop_event = asyncio.Event() async def start(self): + self.db.stop_event.clear() self.advance_loop_task = asyncio.create_task(self.advance()) await self.advance_loop_task self.chain.subscribe() diff --git a/lbry/console.py b/lbry/console.py index 669fb7c9a..e6aead20e 100644 --- a/lbry/console.py +++ b/lbry/console.py @@ -374,11 +374,17 @@ class Advanced(Basic): #self.get_or_create_bar("read", "├─ blocks read", "blocks", d['blocks'], True) #self.get_or_create_bar("save", "└─┬ txs saved", "txs", d['txs'], True) else: - base_name = name[:name.rindex('.')] - for child_name, child_bar in self.bars.items(): - if child_name.startswith(base_name): - child_bar.close() - bar.close() + if d['done'] == (-1,)*len(d['done']): + base_name = name[:name.rindex('.')] + for child_name, child_bar in self.bars.items(): + if child_name.startswith(base_name): + child_bar.close() + bar.close() + else: + if len(d['done']) > 1: + bar.update(d['done']) + else: + bar.update(d['done'][0]) def sync_task(self, name, d): bar_name = f"{name}#{d['id']}" diff --git a/lbry/db/query_context.py b/lbry/db/query_context.py index 09172814f..dfb7a2f7f 100644 --- a/lbry/db/query_context.py +++ b/lbry/db/query_context.py @@ -624,7 +624,7 @@ class BulkLoader: d['expiration_height'] = expiration_height d['takeover_height'] = takeover_height d['is_controlling'] = takeover_height is not None - if d['is_signature_valid']: + if d['is_signature_valid'] and channel_url is not None: d['canonical_url'] = channel_url + '/' + short_url else: d['canonical_url'] = None diff --git a/tests/integration/blockchain/test_blockchain.py b/tests/integration/blockchain/test_blockchain.py index a78ac1d72..b317b6682 100644 --- a/tests/integration/blockchain/test_blockchain.py +++ b/tests/integration/blockchain/test_blockchain.py @@ -741,6 +741,23 @@ class TestGeneralBlockchainSync(SyncingBlockchainTestCase): empty_name, = await self.db.search_claims() self.assertEqual('', empty_name.normalized_name) + async def test_claim_in_abandoned_channel(self): + await self.sync.stop() + channel_1 = await self.get_claim(await self.create_claim(is_channel=True)) + channel_2 = await self.get_claim(await self.create_claim(is_channel=True)) + await self.generate(1, wait=False) + await self.create_claim(sign=channel_1) + await self.create_claim(sign=channel_2) + await self.generate(1, wait=False) + await self.abandon_claim(channel_1.tx_ref.id) + await self.generate(1, wait=False) + await self.sync.start() + c1, c2 = await self.db.search_claims(claim_type='stream') + self.assertEqual(c1.meta['is_signature_valid'], True) # valid at time of pubulish + self.assertIsNone(c1.meta['canonical_url'], None) # channel is abandoned + self.assertEqual(c2.meta['is_signature_valid'], True) + self.assertIsNotNone(c2.meta['canonical_url']) + async def test_short_and_canonical_urls(self): search = self.db.search_claims