diff --git a/lbry/wallet/account.py b/lbry/wallet/account.py index c37457c5d..3a3d4c3f3 100644 --- a/lbry/wallet/account.py +++ b/lbry/wallet/account.py @@ -532,6 +532,9 @@ class Account: return ecdsa.SigningKey.from_pem(private_key_pem, hashfunc=sha256) async def maybe_migrate_certificates(self): + def to_der(private_key_pem): + return ecdsa.SigningKey.from_pem(private_key_pem, hashfunc=sha256).get_verifying_key().to_der() + if not self.channel_keys: return channel_keys = {} @@ -540,8 +543,7 @@ class Account: continue if "-----BEGIN EC PRIVATE KEY-----" not in private_key_pem: continue - private_key = ecdsa.SigningKey.from_pem(private_key_pem, hashfunc=sha256) - public_key_der = private_key.get_verifying_key().to_der() + public_key_der = await asyncio.get_event_loop().run_in_executor(None, to_der, private_key_pem) channel_keys[self.ledger.public_key_to_address(public_key_der)] = private_key_pem if self.channel_keys != channel_keys: self.channel_keys = channel_keys diff --git a/lbry/wallet/ledger.py b/lbry/wallet/ledger.py index 7563e3270..cfd89b6ee 100644 --- a/lbry/wallet/ledger.py +++ b/lbry/wallet/ledger.py @@ -68,12 +68,13 @@ class BlockHeightEvent(NamedTuple): class TransactionCacheItem: - __slots__ = '_tx', 'lock', 'has_tx' + __slots__ = '_tx', 'lock', 'has_tx', 'pending_verifications' def __init__(self, tx: Optional[Transaction] = None, lock: Optional[asyncio.Lock] = None): self.has_tx = asyncio.Event() self.lock = lock or asyncio.Lock() self._tx = self.tx = tx + self.pending_verifications = 0 @property def tx(self) -> Optional[Transaction]: @@ -496,14 +497,15 @@ class Ledger(metaclass=LedgerRegistry): if not we_need: return True - cache_tasks: List[asyncio.Future[Transaction]] = [] + cache_tasks: List[asyncio.Task[Transaction]] = [] synced_history = StringIO() + loop = asyncio.get_running_loop() for i, (txid, remote_height) in enumerate(remote_history): if i < len(local_history) and local_history[i] == (txid, remote_height) and not cache_tasks: synced_history.write(f'{txid}:{remote_height}:') else: check_local = (txid, remote_height) not in we_need - cache_tasks.append(asyncio.ensure_future( + cache_tasks.append(loop.create_task( self.cache_transaction(txid, remote_height, check_local=check_local) )) @@ -579,6 +581,14 @@ class Ledger(metaclass=LedgerRegistry): (cache_item.tx.is_verified or remote_height < 1): return cache_item.tx # cached tx is already up-to-date + try: + cache_item.pending_verifications += 1 + return await self._update_cache_item(cache_item, txid, remote_height, check_local) + finally: + cache_item.pending_verifications -= 1 + + async def _update_cache_item(self, cache_item, txid, remote_height, check_local=True): + async with cache_item.lock: tx = cache_item.tx @@ -587,19 +597,29 @@ class Ledger(metaclass=LedgerRegistry): # check local db tx = cache_item.tx = await self.db.get_transaction(txid=txid) + merkle = None if tx is None: # fetch from network - _raw = await self.network.retriable_call(self.network.get_transaction, txid, remote_height) - tx = Transaction(unhexlify(_raw)) + _raw, merkle = await self.network.retriable_call( + self.network.get_transaction_and_merkle, txid, remote_height + ) + tx = Transaction(unhexlify(_raw), height=merkle.get('block_height')) cache_item.tx = tx # make sure it's saved before caching it - - await self.maybe_verify_transaction(tx, remote_height) + await self.maybe_verify_transaction(tx, remote_height, merkle) return tx - async def maybe_verify_transaction(self, tx, remote_height): + async def maybe_verify_transaction(self, tx, remote_height, merkle=None): tx.height = remote_height - if 0 < remote_height < len(self.headers): - merkle = await self.network.retriable_call(self.network.get_merkle, tx.id, remote_height) + cached = self._tx_cache.get(tx.id) + if not cached: + # cache txs looked up by transaction_show too + cached = TransactionCacheItem() + cached.tx = tx + self._tx_cache[tx.id] = cached + if 0 < remote_height < len(self.headers) and cached.pending_verifications <= 1: + # can't be tx.pending_verifications == 1 because we have to handle the transaction_show case + if not merkle: + merkle = await self.network.retriable_call(self.network.get_merkle, tx.id, remote_height) merkle_root = self.get_root_of_merkle_tree(merkle['merkle'], merkle['pos'], tx.hash) header = await self.headers.get(remote_height) tx.position = merkle['pos'] diff --git a/lbry/wallet/manager.py b/lbry/wallet/manager.py index 37b2ec992..20658a2e5 100644 --- a/lbry/wallet/manager.py +++ b/lbry/wallet/manager.py @@ -256,18 +256,20 @@ class WalletManager: def get_unused_address(self): return self.default_account.receiving.get_or_create_usable_address() - async def get_transaction(self, txid): + async def get_transaction(self, txid: str): tx = await self.db.get_transaction(txid=txid) - if not tx: - try: - raw = await self.ledger.network.get_transaction(txid) - height = await self.ledger.network.get_transaction_height(txid) - except CodeMessageError as e: - if 'No such mempool or blockchain transaction.' in e.message: - return {'success': False, 'code': 404, 'message': 'transaction not found'} - return {'success': False, 'code': e.code, 'message': e.message} - tx = Transaction(unhexlify(raw)) - await self.ledger.maybe_verify_transaction(tx, height) + if tx: + return tx + try: + raw, merkle = await self.ledger.network.get_transaction_and_merkle(txid) + except CodeMessageError as e: + if 'No such mempool or blockchain transaction.' in e.message: + return {'success': False, 'code': 404, 'message': 'transaction not found'} + return {'success': False, 'code': e.code, 'message': e.message} + height = merkle.get('block_height') + tx = Transaction(unhexlify(raw), height=height) + if height and height > 0: + await self.ledger.maybe_verify_transaction(tx, height, merkle) return tx async def create_purchase_transaction( diff --git a/lbry/wallet/network.py b/lbry/wallet/network.py index f0739fa1c..b117a0164 100644 --- a/lbry/wallet/network.py +++ b/lbry/wallet/network.py @@ -55,7 +55,7 @@ class ClientSession(BaseClientSession): async def send_request(self, method, args=()): self.pending_amount += 1 - log.debug("send %s to %s:%i", method, *self.server) + log.debug("send %s%s to %s:%i", method, tuple(args), *self.server) try: if method == 'server.version': return await self.send_timed_server_version_request(args, self.timeout) @@ -156,7 +156,7 @@ class ClientSession(BaseClientSession): class Network: PROTOCOL_VERSION = __version__ - MINIMUM_REQUIRED = (0, 59, 0) + MINIMUM_REQUIRED = (0, 65, 0) def __init__(self, ledger): self.ledger = ledger @@ -256,6 +256,11 @@ class Network: restricted = known_height in (None, -1, 0) or 0 > known_height > self.remote_height - 10 return self.rpc('blockchain.transaction.get', [tx_hash], restricted) + def get_transaction_and_merkle(self, tx_hash, known_height=None): + # use any server if its old, otherwise restrict to who gave us the history + restricted = known_height in (None, -1, 0) or 0 > known_height > self.remote_height - 10 + return self.rpc('blockchain.transaction.info', [tx_hash], restricted) + def get_transaction_height(self, tx_hash, known_height=None): restricted = not known_height or 0 > known_height > self.remote_height - 10 return self.rpc('blockchain.transaction.get_height', [tx_hash], restricted) diff --git a/tests/unit/wallet/test_ledger.py b/tests/unit/wallet/test_ledger.py index dc51ca240..0244de987 100644 --- a/tests/unit/wallet/test_ledger.py +++ b/tests/unit/wallet/test_ledger.py @@ -33,6 +33,13 @@ class MockNetwork: self.get_transaction_called.append(tx_hash) return self.transaction[tx_hash] + async def get_transaction_and_merkle(self, tx_hash, known_height=None): + tx = await self.get_transaction(tx_hash) + merkle = {} + if known_height: + merkle = await self.get_merkle(tx_hash, known_height) + return tx, merkle + class LedgerTestCase(AsyncioTestCase): @@ -73,6 +80,11 @@ class LedgerTestCase(AsyncioTestCase): class TestSynchronization(LedgerTestCase): async def test_update_history(self): + txid1 = '252bda9b22cc902ca2aa2de3548ee8baf06b8501ff7bfb3b0b7d980dbd1bf792' + txid2 = 'ab9c0654dd484ac20437030f2034e25dcb29fc507e84b91138f80adc3af738f9' + txid3 = 'a2ae3d1db3c727e7d696122cab39ee20a7f81856dab7019056dd539f38c548a0' + txid4 = '047cf1d53ef68f0fd586d46f90c09ff8e57a4180f67e7f4b8dd0135c3741e828' + account = Account.generate(self.ledger, Wallet(), "torba") address = await account.receiving.get_or_create_usable_address() address_details = await self.ledger.db.get_address(address=address) @@ -83,46 +95,49 @@ class TestSynchronization(LedgerTestCase): self.add_header(block_height=2, merkle_root=b'abcd04') self.add_header(block_height=3, merkle_root=b'abcd04') self.ledger.network = MockNetwork([ - {'tx_hash': 'abcd01', 'height': 0}, - {'tx_hash': 'abcd02', 'height': 1}, - {'tx_hash': 'abcd03', 'height': 2}, + {'tx_hash': txid1, 'height': 0}, + {'tx_hash': txid2, 'height': 1}, + {'tx_hash': txid3, 'height': 2}, ], { - 'abcd01': hexlify(get_transaction(get_output(1)).raw), - 'abcd02': hexlify(get_transaction(get_output(2)).raw), - 'abcd03': hexlify(get_transaction(get_output(3)).raw), + txid1: hexlify(get_transaction(get_output(1)).raw), + txid2: hexlify(get_transaction(get_output(2)).raw), + txid3: hexlify(get_transaction(get_output(3)).raw), }) await self.ledger.update_history(address, '') self.assertListEqual(self.ledger.network.get_history_called, [address]) - self.assertListEqual(self.ledger.network.get_transaction_called, ['abcd01', 'abcd02', 'abcd03']) + self.assertListEqual(self.ledger.network.get_transaction_called, [txid1, txid2, txid3]) address_details = await self.ledger.db.get_address(address=address) + self.assertEqual( address_details['history'], - '252bda9b22cc902ca2aa2de3548ee8baf06b8501ff7bfb3b0b7d980dbd1bf792:0:' - 'ab9c0654dd484ac20437030f2034e25dcb29fc507e84b91138f80adc3af738f9:1:' - 'a2ae3d1db3c727e7d696122cab39ee20a7f81856dab7019056dd539f38c548a0:2:' + f'{txid1}:0:' + f'{txid2}:1:' + f'{txid3}:2:' ) self.ledger.network.get_history_called = [] self.ledger.network.get_transaction_called = [] + for cache_item in self.ledger._tx_cache.values(): + cache_item.tx.is_verified = True await self.ledger.update_history(address, '') self.assertListEqual(self.ledger.network.get_history_called, [address]) self.assertListEqual(self.ledger.network.get_transaction_called, []) - self.ledger.network.history.append({'tx_hash': 'abcd04', 'height': 3}) - self.ledger.network.transaction['abcd04'] = hexlify(get_transaction(get_output(4)).raw) + self.ledger.network.history.append({'tx_hash': txid4, 'height': 3}) + self.ledger.network.transaction[txid4] = hexlify(get_transaction(get_output(4)).raw) self.ledger.network.get_history_called = [] self.ledger.network.get_transaction_called = [] await self.ledger.update_history(address, '') self.assertListEqual(self.ledger.network.get_history_called, [address]) - self.assertListEqual(self.ledger.network.get_transaction_called, ['abcd04']) + self.assertListEqual(self.ledger.network.get_transaction_called, [txid4]) address_details = await self.ledger.db.get_address(address=address) self.assertEqual( address_details['history'], - '252bda9b22cc902ca2aa2de3548ee8baf06b8501ff7bfb3b0b7d980dbd1bf792:0:' - 'ab9c0654dd484ac20437030f2034e25dcb29fc507e84b91138f80adc3af738f9:1:' - 'a2ae3d1db3c727e7d696122cab39ee20a7f81856dab7019056dd539f38c548a0:2:' - '047cf1d53ef68f0fd586d46f90c09ff8e57a4180f67e7f4b8dd0135c3741e828:3:' + f'{txid1}:0:' + f'{txid2}:1:' + f'{txid3}:2:' + f'{txid4}:3:' )