test_reorg_dropping_claim

lbry/wallet/orchstr8/node.py

@@ -55,7 +55,8 @@ class Conductor:
 
     async def start_blockchain(self):
         if not self.blockchain_started:
-            await self.blockchain_node.start()
+            asyncio.create_task(self.blockchain_node.start())
+            await self.blockchain_node.running.wait()
             await self.blockchain_node.generate(200)
             self.blockchain_started = True
 
@@ -255,6 +256,10 @@ class BlockchainNode:
         self.rpcport = 9245 + 2  # avoid conflict with default rpc port
         self.rpcuser = 'rpcuser'
         self.rpcpassword = 'rpcpassword'
+        self.stopped = False
+        self.restart_ready = asyncio.Event()
+        self.restart_ready.set()
+        self.running = asyncio.Event()
 
     @property
     def rpc_url(self):
@@ -315,13 +320,27 @@ class BlockchainNode:
             f'-port={self.peerport}'
         ]
         self.log.info(' '.join(command))
-        self.transport, self.protocol = await loop.subprocess_exec(
+        while not self.stopped:
-            BlockchainProcess, *command
+            if self.running.is_set():
-        )
+                await asyncio.sleep(1)
-        await self.protocol.ready.wait()
+                continue
-        assert not self.protocol.stopped.is_set()
+            await self.restart_ready.wait()
+            try:
+                self.transport, self.protocol = await loop.subprocess_exec(
+                    BlockchainProcess, *command
+                )
+                await self.protocol.ready.wait()
+                assert not self.protocol.stopped.is_set()
+                self.running.set()
+            except asyncio.CancelledError:
+                self.running.clear()
+                raise
+            except:
+                self.running.clear()
+                log.exception("boom")
 
     async def stop(self, cleanup=True):
+        self.stopped = True
         try:
             self.transport.terminate()
             await self.protocol.stopped.wait()
@@ -330,6 +349,16 @@ class BlockchainNode:
             if cleanup:
                 self.cleanup()
 
+    async def clear_mempool(self):
+        self.restart_ready.clear()
+        self.transport.terminate()
+        await self.protocol.stopped.wait()
+        self.transport.close()
+        self.running.clear()
+        os.remove(os.path.join(self.data_path, 'regtest', 'mempool.dat'))
+        self.restart_ready.set()
+        await self.running.wait()
+
     def cleanup(self):
         shutil.rmtree(self.data_path, ignore_errors=True)
 
@@ -361,6 +390,9 @@ class BlockchainNode:
     def get_block_hash(self, block):
         return self._cli_cmnd('getblockhash', str(block))
 
+    async def get_block(self, block_hash):
+        return json.loads(await self._cli_cmnd('getblock', block_hash, '1'))
+
     def get_raw_change_address(self):
         return self._cli_cmnd('getrawchangeaddress')
 

tests/integration/blockchain/test_blockchain_reorganization.py

@@ -1,6 +1,9 @@
 import logging
+import asyncio
+from binascii import unhexlify
 from lbry.testcase import CommandTestCase
 from lbry.wallet.server.prometheus import REORG_COUNT
+from lbry.wallet.transaction import Transaction
 
 
 class BlockchainReorganizationTests(CommandTestCase):
@@ -35,3 +38,44 @@ class BlockchainReorganizationTests(CommandTestCase):
         await self.assertBlockHash(207)
         await self.assertBlockHash(208)
         self.assertEqual(2, REORG_COUNT._samples()[0][2])
+
+    async def test_reorg_dropping_claim(self):
+        # sanity check
+        result_txs, _, _, _ = await self.ledger.claim_search([], name='hovercraft')
+        self.assertListEqual(result_txs, [])
+
+        # create a claim and verify it is returned by claim_search
+        self.assertEqual(self.ledger.headers.height, 206)
+        broadcast_tx = Transaction(unhexlify((await self.stream_create(name='hovercraft'))['hex'].encode()))
+        self.assertEqual(self.ledger.headers.height, 207)
+        result_txs, _, _, _ = await self.ledger.claim_search([], name='hovercraft')
+        self.assertEqual(1, len(result_txs))
+        tx = result_txs[0]
+        self.assertEqual(tx.tx_ref.id, broadcast_tx.id)
+
+        # check that our tx is in block 207 as returned by lbrycrdd
+        invalidated_block_hash = (await self.ledger.headers.hash(207)).decode()
+        block_207 = await self.blockchain.get_block(invalidated_block_hash)
+        self.assertIn(tx.tx_ref.id, block_207['tx'])
+
+        # reorg the last block dropping our claim tx
+        await self.blockchain.invalidate_block(invalidated_block_hash)
+        await self.blockchain.clear_mempool()
+        await self.blockchain.generate(2)
+
+        # verify the claim was dropped from block 207 as returned by lbrycrdd
+        reorg_block_hash = await self.blockchain.get_block_hash(207)
+        self.assertNotEqual(invalidated_block_hash, reorg_block_hash)
+        block_207 = await self.blockchain.get_block(reorg_block_hash)
+        self.assertNotIn(tx.tx_ref.id, block_207['tx'])
+
+        # wait for the client to catch up and verify the reorg
+        await asyncio.wait_for(self.on_header(208), 3.0)
+        await self.assertBlockHash(206)
+        await self.assertBlockHash(207)
+        await self.assertBlockHash(208)
+        client_reorg_block_hash = (await self.ledger.headers.hash(207)).decode()
+        self.assertEqual(client_reorg_block_hash, reorg_block_hash)
+
+        result_txs, _, _, _ = await self.ledger.claim_search([], name='hovercraft')
+        self.assertListEqual(result_txs, [])