Madiator2011/lbry-sdk
1
0
Fork 0
forked from LBRYCommunity/lbry-sdk
Code Pull requests Activity

Merge pull request #2852 from lbryio/faster-transaction-sync

Browse source 
More efficient syncing of wallet transactions
This commit is contained in:
Jack Robison 2020-03-26 11:43:20 -04:00 • committed by GitHub
commit fdb42ac876
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 86 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
lbry/wallet/account.py
View file

@ -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

38
lbry/wallet/ledger.py
View file

@ -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,18 +597,28 @@ 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):
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)

14
lbry/wallet/manager.py
View file

@ -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:
if tx:
return tx
try:
raw = await self.ledger.network.get_transaction(txid)
height = await self.ledger.network.get_transaction_height(txid)
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}
tx = Transaction(unhexlify(raw))
await self.ledger.maybe_verify_transaction(tx, height)
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(

9
lbry/wallet/network.py
View file

@ -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)

49
tests/unit/wallet/test_ledger.py
View file

@ -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:'
)