Merge pull request #3588 from lbryio/scribe

move `lbry.wallet.server` to new project called `scribe`, switch from using `lbrycrd` to `lbcd` in integration tests

lbry/extras/daemon/daemon.py

@@ -53,8 +53,7 @@
 from lbry.file_analysis import VideoFileAnalyzer
 from lbry.schema.claim import Claim
 from lbry.schema.url import URL, normalize_name
-from lbry.wallet.server.db.elasticsearch.constants import RANGE_FIELDS, REPLACEMENTS
-MY_RANGE_FIELDS = RANGE_FIELDS - {"limit_claims_per_channel"}
+
 
 if typing.TYPE_CHECKING:
     from lbry.blob.blob_manager import BlobManager
@@ -67,6 +66,29 @@
 
 log = logging.getLogger(__name__)
 
+RANGE_FIELDS = {
+    'height', 'creation_height', 'activation_height', 'expiration_height',
+    'timestamp', 'creation_timestamp', 'duration', 'release_time', 'fee_amount',
+    'tx_position', 'repost_count', 'limit_claims_per_channel',
+    'amount', 'effective_amount', 'support_amount',
+    'trending_score', 'censor_type', 'tx_num'
+}
+MY_RANGE_FIELDS = RANGE_FIELDS - {"limit_claims_per_channel"}
+REPLACEMENTS = {
+    'claim_name': 'normalized_name',
+    'name': 'normalized_name',
+    'txid': 'tx_id',
+    'nout': 'tx_nout',
+    'trending_group': 'trending_score',
+    'trending_mixed': 'trending_score',
+    'trending_global': 'trending_score',
+    'trending_local': 'trending_score',
+    'reposted': 'repost_count',
+    'stream_types': 'stream_type',
+    'media_types': 'media_type',
+    'valid_channel_signature': 'is_signature_valid'
+}
+
 
 def is_transactional_function(name):
     for action in ('create', 'update', 'abandon', 'send', 'fund'):

lbry/schema/result.py

@@ -1,13 +1,11 @@
 import base64
-from typing import List, TYPE_CHECKING, Union, Optional
+from typing import List, Union, Optional, NamedTuple
 from binascii import hexlify
 from itertools import chain
 
 from lbry.error import ResolveCensoredError
 from lbry.schema.types.v2.result_pb2 import Outputs as OutputsMessage
 from lbry.schema.types.v2.result_pb2 import Error as ErrorMessage
-if TYPE_CHECKING:
-    from lbry.wallet.server.leveldb import ResolveResult
 
 INVALID = ErrorMessage.Code.Name(ErrorMessage.INVALID)
 NOT_FOUND = ErrorMessage.Code.Name(ErrorMessage.NOT_FOUND)
@@ -24,6 +22,31 @@ def set_reference(reference, claim_hash, rows):
                 return
 
 
+class ResolveResult(NamedTuple):
+    name: str
+    normalized_name: str
+    claim_hash: bytes
+    tx_num: int
+    position: int
+    tx_hash: bytes
+    height: int
+    amount: int
+    short_url: str
+    is_controlling: bool
+    canonical_url: str
+    creation_height: int
+    activation_height: int
+    expiration_height: int
+    effective_amount: int
+    support_amount: int
+    reposted: int
+    last_takeover_height: Optional[int]
+    claims_in_channel: Optional[int]
+    channel_hash: Optional[bytes]
+    reposted_claim_hash: Optional[bytes]
+    signature_valid: Optional[bool]
+
+
 class Censor:
 
     NOT_CENSORED = 0

lbry/testcase.py

@@ -19,7 +19,7 @@
 from lbry.wallet.util import satoshis_to_coins
 from lbry.wallet.dewies import lbc_to_dewies
 from lbry.wallet.orchstr8 import Conductor
-from lbry.wallet.orchstr8.node import BlockchainNode, WalletNode, HubNode
+from lbry.wallet.orchstr8.node import LBCWalletNode, WalletNode, HubNode
 from lbry.schema.claim import Claim
 
 from lbry.extras.daemon.daemon import Daemon, jsonrpc_dumps_pretty
@@ -236,7 +236,7 @@ class IntegrationTestCase(AsyncioTestCase):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.conductor: Optional[Conductor] = None
-        self.blockchain: Optional[BlockchainNode] = None
+        self.blockchain: Optional[LBCWalletNode] = None
         self.hub: Optional[HubNode] = None
         self.wallet_node: Optional[WalletNode] = None
         self.manager: Optional[WalletManager] = None
@@ -246,15 +246,17 @@ def __init__(self, *args, **kwargs):
 
     async def asyncSetUp(self):
         self.conductor = Conductor(seed=self.SEED)
-        await self.conductor.start_blockchain()
-        self.addCleanup(self.conductor.stop_blockchain)
+        await self.conductor.start_lbcd()
+        self.addCleanup(self.conductor.stop_lbcd)
+        await self.conductor.start_lbcwallet()
+        self.addCleanup(self.conductor.stop_lbcwallet)
         await self.conductor.start_spv()
         self.addCleanup(self.conductor.stop_spv)
         await self.conductor.start_wallet()
         self.addCleanup(self.conductor.stop_wallet)
         await self.conductor.start_hub()
         self.addCleanup(self.conductor.stop_hub)
-        self.blockchain = self.conductor.blockchain_node
+        self.blockchain = self.conductor.lbcwallet_node
         self.hub = self.conductor.hub_node
         self.wallet_node = self.conductor.wallet_node
         self.manager = self.wallet_node.manager
@@ -269,18 +271,50 @@ async def assertBalance(self, account, expected_balance: str):  # pylint: disabl
     def broadcast(self, tx):
         return self.ledger.broadcast(tx)
 
+    async def broadcast_and_confirm(self, tx, ledger=None):
+        ledger = ledger or self.ledger
+        notifications = asyncio.create_task(ledger.wait(tx))
+        await ledger.broadcast(tx)
+        await notifications
+        await self.generate_and_wait(1, [tx.id], ledger)
+
     async def on_header(self, height):
         if self.ledger.headers.height < height:
             await self.ledger.on_header.where(
                 lambda e: e.height == height
             )
         return True
 
-    def on_transaction_id(self, txid, ledger=None):
-        return (ledger or self.ledger).on_transaction.where(
-            lambda e: e.tx.id == txid
+    async def send_to_address_and_wait(self, address, amount, blocks_to_generate=0, ledger=None):
+        tx_watch = []
+        txid = None
+        done = False
+        watcher = (ledger or self.ledger).on_transaction.where(
+            lambda e: e.tx.id == txid or done or tx_watch.append(e.tx.id)
         )
 
+        txid = await self.blockchain.send_to_address(address, amount)
+        done = txid in tx_watch
+        await watcher
+
+        await self.generate_and_wait(blocks_to_generate, [txid], ledger)
+        return txid
+
+    async def generate_and_wait(self, blocks_to_generate, txids, ledger=None):
+        if blocks_to_generate > 0:
+            watcher = (ledger or self.ledger).on_transaction.where(
+                lambda e: ((e.tx.id in txids and txids.remove(e.tx.id)), len(txids) <= 0)[-1]  # multi-statement lambda
+            )
+            self.conductor.spv_node.server.synchronized.clear()
+            await self.blockchain.generate(blocks_to_generate)
+            height = self.blockchain.block_expected
+            await watcher
+            while True:
+                await self.conductor.spv_node.server.synchronized.wait()
+                self.conductor.spv_node.server.synchronized.clear()
+                if self.conductor.spv_node.server.db.db_height >= height:
+                    break
+
     def on_address_update(self, address):
         return self.ledger.on_transaction.where(
             lambda e: e.address == address
@@ -291,6 +325,19 @@ def on_transaction_address(self, tx, address):
             lambda e: e.tx.id == tx.id and e.address == address
         )
 
+    async def generate(self, blocks):
+        """ Ask lbrycrd to generate some blocks and wait until ledger has them. """
+        prepare = self.ledger.on_header.where(self.blockchain.is_expected_block)
+        height = self.blockchain.block_expected
+        self.conductor.spv_node.server.synchronized.clear()
+        await self.blockchain.generate(blocks)
+        await prepare  # no guarantee that it didn't happen already, so start waiting from before calling generate
+        while True:
+            await self.conductor.spv_node.server.synchronized.wait()
+            self.conductor.spv_node.server.synchronized.clear()
+            if self.conductor.spv_node.server.db.db_height >= height:
+                break
+
 
 class FakeExchangeRateManager(ExchangeRateManager):
 
@@ -351,20 +398,19 @@ def __init__(self, *args, **kwargs):
         self.skip_libtorrent = True
 
     async def asyncSetUp(self):
-        await super().asyncSetUp()
 
         logging.getLogger('lbry.blob_exchange').setLevel(self.VERBOSITY)
         logging.getLogger('lbry.daemon').setLevel(self.VERBOSITY)
         logging.getLogger('lbry.stream').setLevel(self.VERBOSITY)
         logging.getLogger('lbry.wallet').setLevel(self.VERBOSITY)
 
+        await super().asyncSetUp()
+
         self.daemon = await self.add_daemon(self.wallet_node)
 
         await self.account.ensure_address_gap()
         address = (await self.account.receiving.get_addresses(limit=1, only_usable=True))[0]
-        sendtxid = await self.blockchain.send_to_address(address, 10)
-        await self.confirm_tx(sendtxid)
-        await self.generate(5)
+        await self.send_to_address_and_wait(address, 10, 6)
 
         server_tmp_dir = tempfile.mkdtemp()
         self.addCleanup(shutil.rmtree, server_tmp_dir)
@@ -461,9 +507,14 @@ def wallet_maker(component_manager):
 
     async def confirm_tx(self, txid, ledger=None):
         """ Wait for tx to be in mempool, then generate a block, wait for tx to be in a block. """
-        await self.on_transaction_id(txid, ledger)
-        await self.generate(1)
-        await self.on_transaction_id(txid, ledger)
+        # await (ledger or self.ledger).on_transaction.where(lambda e: e.tx.id == txid)
+        on_tx = (ledger or self.ledger).on_transaction.where(lambda e: e.tx.id == txid)
+        await asyncio.wait([self.generate(1), on_tx], timeout=5)
+
+        # # actually, if it's in the mempool or in the block we're fine
+        # await self.generate_and_wait(1, [txid], ledger=ledger)
+        # return txid
+
         return txid
 
     async def on_transaction_dict(self, tx):
@@ -478,12 +529,6 @@ def get_all_addresses(tx):
             addresses.add(txo['address'])
         return list(addresses)
 
-    async def generate(self, blocks):
-        """ Ask lbrycrd to generate some blocks and wait until ledger has them. """
-        prepare = self.ledger.on_header.where(self.blockchain.is_expected_block)
-        await self.blockchain.generate(blocks)
-        await prepare  # no guarantee that it didn't happen already, so start waiting from before calling generate
-
     async def blockchain_claim_name(self, name: str, value: str, amount: str, confirm=True):
         txid = await self.blockchain._cli_cmnd('claimname', name, value, amount)
         if confirm:
@@ -514,7 +559,7 @@ async def confirm_and_render(self, awaitable, confirm, return_tx=False) -> Trans
             return self.sout(tx)
         return tx
 
-    async def create_nondeterministic_channel(self, name, price, pubkey_bytes, daemon=None):
+    async def create_nondeterministic_channel(self, name, price, pubkey_bytes, daemon=None, blocking=False):
         account = (daemon or self.daemon).wallet_manager.default_account
         claim_address = await account.receiving.get_or_create_usable_address()
         claim = Claim()
@@ -524,7 +569,7 @@ async def create_nondeterministic_channel(self, name, price, pubkey_bytes, daemo
             claim_address, [self.account], self.account
         )
         await tx.sign([self.account])
-        await (daemon or self.daemon).broadcast_or_release(tx, False)
+        await (daemon or self.daemon).broadcast_or_release(tx, blocking)
         return self.sout(tx)
 
     def create_upload_file(self, data, prefix=None, suffix=None):

lbry/utils.py

@@ -405,7 +405,7 @@ async def fallback_get_external_ip():  # used if spv servers can't be used for i
 
 async def _get_external_ip(default_servers) -> typing.Tuple[typing.Optional[str], typing.Optional[str]]:
     # used if upnp is disabled or non-functioning
-    from lbry.wallet.server.udp import SPVStatusClientProtocol  # pylint: disable=C0415
+    from lbry.wallet.udp import SPVStatusClientProtocol  # pylint: disable=C0415
 
     hostname_to_ip = {}
     ip_to_hostnames = collections.defaultdict(list)

lbry/wallet/__init__.py

@@ -1,17 +1,23 @@
-__node_daemon__ = 'lbrycrdd'
-__node_cli__ = 'lbrycrd-cli'
-__node_bin__ = ''
-__node_url__ = (
-    'https://github.com/lbryio/lbrycrd/releases/download/v0.17.4.6/lbrycrd-linux-1746.zip'
+__lbcd__ = 'lbcd'
+__lbcctl__ = 'lbcctl'
+__lbcwallet__ = 'lbcwallet'
+__lbcd_url__ = (
+    'https://github.com/lbryio/lbcd/releases/download/' +
+    'v0.22.200-beta/lbcd_0.22.200-beta_TARGET_PLATFORM.tar.gz'
+)
+__lbcwallet_url__ = (
+    'https://github.com/lbryio/lbcwallet/releases/download/' +
+    'v0.13.100-alpha-rc2/lbcwallet_0.13.100-alpha-rc2_TARGET_PLATFORM.tar.gz'
 )
 __spvserver__ = 'lbry.wallet.server.coin.LBCRegTest'
 
-from .wallet import Wallet, WalletStorage, TimestampedPreferences, ENCRYPT_ON_DISK
-from .manager import WalletManager
-from .network import Network
-from .ledger import Ledger, RegTestLedger, TestNetLedger, BlockHeightEvent
-from .account import Account, AddressManager, SingleKey, HierarchicalDeterministic, DeterministicChannelKeyManager
-from .transaction import Transaction, Output, Input
-from .script import OutputScript, InputScript
-from .database import SQLiteMixin, Database
-from .header import Headers
+from lbry.wallet.wallet import Wallet, WalletStorage, TimestampedPreferences, ENCRYPT_ON_DISK
+from lbry.wallet.manager import WalletManager
+from lbry.wallet.network import Network
+from lbry.wallet.ledger import Ledger, RegTestLedger, TestNetLedger, BlockHeightEvent
+from lbry.wallet.account import Account, AddressManager, SingleKey, HierarchicalDeterministic, \
+    DeterministicChannelKeyManager
+from lbry.wallet.transaction import Transaction, Output, Input
+from lbry.wallet.script import OutputScript, InputScript
+from lbry.wallet.database import SQLiteMixin, Database
+from lbry.wallet.header import Headers

lbry/wallet/ledger.py

@@ -16,18 +16,18 @@
 from lbry.crypto.base58 import Base58
 from lbry.utils import LRUCacheWithMetrics
 
-from .tasks import TaskGroup
-from .database import Database
-from .stream import StreamController
-from .dewies import dewies_to_lbc
-from .account import Account, AddressManager, SingleKey
-from .network import Network
-from .transaction import Transaction, Output
-from .header import Headers, UnvalidatedHeaders
-from .checkpoints import HASHES
-from .constants import TXO_TYPES, CLAIM_TYPES, COIN, NULL_HASH32
-from .bip32 import PublicKey, PrivateKey
-from .coinselection import CoinSelector
+from lbry.wallet.tasks import TaskGroup
+from lbry.wallet.database import Database
+from lbry.wallet.stream import StreamController
+from lbry.wallet.dewies import dewies_to_lbc
+from lbry.wallet.account import Account, AddressManager, SingleKey
+from lbry.wallet.network import Network
+from lbry.wallet.transaction import Transaction, Output
+from lbry.wallet.header import Headers, UnvalidatedHeaders
+from lbry.wallet.checkpoints import HASHES
+from lbry.wallet.constants import TXO_TYPES, CLAIM_TYPES, COIN, NULL_HASH32
+from lbry.wallet.bip32 import PublicKey, PrivateKey
+from lbry.wallet.coinselection import CoinSelector
 
 log = logging.getLogger(__name__)
 
@@ -365,6 +365,10 @@ async def stop(self):
         await self.db.close()
         await self.headers.close()
 
+    async def tasks_are_done(self):
+        await self._update_tasks.done.wait()
+        await self._other_tasks.done.wait()
+
     @property
     def local_height_including_downloaded_height(self):
         return max(self.headers.height, self._download_height)
@@ -739,7 +743,7 @@ async def wait(self, tx: Transaction, height=-1, timeout=1):
         while timeout and (int(time.perf_counter()) - start) <= timeout:
             if await self._wait_round(tx, height, addresses):
                 return
-        raise asyncio.TimeoutError('Timed out waiting for transaction.')
+        raise asyncio.TimeoutError(f'Timed out waiting for transaction. {tx.id}')
 
     async def _wait_round(self, tx: Transaction, height: int, addresses: Iterable[str]):
         records = await self.db.get_addresses(address__in=addresses)
@@ -782,7 +786,7 @@ async def _inflate_outputs(
         if hub_server:
             outputs = Outputs.from_grpc(encoded_outputs)
         else:
-            outputs = Outputs.from_base64(encoded_outputs or b'')  # TODO: why is the server returning None?
+            outputs = Outputs.from_base64(encoded_outputs or '')  # TODO: why is the server returning None?
         txs: List[Transaction] = []
         if len(outputs.txs) > 0:
             async for tx in self.request_transactions(tuple(outputs.txs), cached=True):