diff --git a/lbry/blob_exchange/server.py b/lbry/blob_exchange/server.py index 6e4950eea..8ee9212ff 100644 --- a/lbry/blob_exchange/server.py +++ b/lbry/blob_exchange/server.py @@ -138,7 +138,7 @@ class BlobServerProtocol(asyncio.Protocol): try: request = BlobRequest.deserialize(self.buf + data) self.buf = remainder - except JSONDecodeError: + except (UnicodeDecodeError, JSONDecodeError): log.error("request from %s is not valid json (%i bytes): %s", self.peer_address_and_port, len(self.buf + data), '' if not data else binascii.hexlify(self.buf + data).decode()) self.close() diff --git a/lbry/wallet/orchstr8/node.py b/lbry/wallet/orchstr8/node.py index a706ec9c9..1f0fd1bca 100644 --- a/lbry/wallet/orchstr8/node.py +++ b/lbry/wallet/orchstr8/node.py @@ -18,14 +18,6 @@ import urllib.request from uuid import uuid4 -try: - from scribe.hub.env import ServerEnv - from scribe.hub.service import HubServerService - from scribe.elasticsearch.service import ElasticSyncService - from scribe.blockchain.service import BlockchainProcessorService -except ImportError: - pass - import lbry from lbry.wallet import Wallet, Ledger, RegTestLedger, WalletManager, Account, BlockHeightEvent from lbry.conf import KnownHubsList, Config @@ -33,6 +25,16 @@ from lbry.wallet.orchstr8 import __hub_url__ log = logging.getLogger(__name__) +try: + from hub.herald.env import ServerEnv + from hub.scribe.env import BlockchainEnv + from hub.elastic_sync.env import ElasticEnv + from hub.herald.service import HubServerService + from hub.elastic_sync.service import ElasticSyncService + from hub.scribe.service import BlockchainProcessorService +except ImportError: + pass + def get_lbcd_node_from_ledger(ledger_module): return LBCDNode( @@ -257,14 +259,24 @@ class SPVNode: 'session_timeout': self.session_timeout, 'max_query_workers': 0, 'es_index_prefix': self.index_name, - 'chain': 'regtest' + 'chain': 'regtest', + 'index_address_status': False } if extraconf: conf.update(extraconf) - env = ServerEnv(**conf) - self.writer = BlockchainProcessorService(env) - self.server = HubServerService(env) - self.es_writer = ElasticSyncService(env) + self.writer = BlockchainProcessorService( + BlockchainEnv(db_dir=self.data_path, daemon_url=lbcwallet_node.rpc_url, + reorg_limit=100, max_query_workers=0, chain='regtest', index_address_status=False) + ) + self.server = HubServerService(ServerEnv(**conf)) + self.es_writer = ElasticSyncService( + ElasticEnv( + db_dir=self.data_path, reorg_limit=100, max_query_workers=0, chain='regtest', + elastic_notifier_port=self.elastic_notifier_port, + es_index_prefix=self.index_name, filtering_channel_ids=(extraconf or {}).get('filtering_channel_ids'), + blocking_channel_ids=(extraconf or {}).get('blocking_channel_ids') + ) + ) await self.writer.start() await self.es_writer.start() await self.server.start() diff --git a/tests/integration/blockchain/test_blockchain_reorganization.py b/tests/integration/blockchain/test_blockchain_reorganization.py index 10ae90e6e..a48705c92 100644 --- a/tests/integration/blockchain/test_blockchain_reorganization.py +++ b/tests/integration/blockchain/test_blockchain_reorganization.py @@ -10,20 +10,21 @@ class BlockchainReorganizationTests(CommandTestCase): async def assertBlockHash(self, height): bp = self.conductor.spv_node.writer + reader = self.conductor.spv_node.server def get_txids(): return [ - bp.db.fs_tx_hash(tx_num)[0][::-1].hex() + reader.db.fs_tx_hash(tx_num)[0][::-1].hex() for tx_num in range(bp.db.tx_counts[height - 1], bp.db.tx_counts[height]) ] block_hash = await self.blockchain.get_block_hash(height) self.assertEqual(block_hash, (await self.ledger.headers.hash(height)).decode()) - self.assertEqual(block_hash, (await bp.db.fs_block_hashes(height, 1))[0][::-1].hex()) + self.assertEqual(block_hash, (await reader.db.fs_block_hashes(height, 1))[0][::-1].hex()) txids = await asyncio.get_event_loop().run_in_executor(None, get_txids) - txs = await bp.db.get_transactions_and_merkles(txids) + txs = await reader.db.get_transactions_and_merkles(txids) block_txs = (await bp.daemon.deserialised_block(block_hash))['tx'] self.assertSetEqual(set(block_txs), set(txs.keys()), msg='leveldb/lbrycrd is missing transactions') self.assertListEqual(block_txs, list(txs.keys()), msg='leveldb/lbrycrd transactions are of order') diff --git a/tests/integration/blockchain/test_network.py b/tests/integration/blockchain/test_network.py index 230110596..24cd5e8f4 100644 --- a/tests/integration/blockchain/test_network.py +++ b/tests/integration/blockchain/test_network.py @@ -1,12 +1,12 @@ import asyncio -import scribe +import hub from unittest.mock import Mock -from scribe.hub import HUB_PROTOCOL_VERSION -from scribe.hub.udp import StatusServer -from scribe.hub.session import LBRYElectrumX -from scribe.blockchain.network import LBCRegTest +from hub.herald import HUB_PROTOCOL_VERSION +from hub.herald.udp import StatusServer +from hub.herald.session import LBRYElectrumX +from hub.scribe.network import LBCRegTest from lbry.wallet.network import Network from lbry.wallet.orchstr8 import Conductor diff --git a/tests/integration/blockchain/test_wallet_server_sessions.py b/tests/integration/blockchain/test_wallet_server_sessions.py index d09c398f7..5cc7b3902 100644 --- a/tests/integration/blockchain/test_wallet_server_sessions.py +++ b/tests/integration/blockchain/test_wallet_server_sessions.py @@ -1,7 +1,7 @@ import asyncio -from scribe.hub import HUB_PROTOCOL_VERSION -from scribe.hub.session import LBRYElectrumX +from hub.herald import HUB_PROTOCOL_VERSION +from hub.herald.session import LBRYElectrumX from lbry.error import ServerPaymentFeeAboveMaxAllowedError from lbry.wallet.network import ClientSession diff --git a/tests/integration/takeovers/test_resolve_command.py b/tests/integration/takeovers/test_resolve_command.py index ebef0f917..5ffe18224 100644 --- a/tests/integration/takeovers/test_resolve_command.py +++ b/tests/integration/takeovers/test_resolve_command.py @@ -326,6 +326,30 @@ class ResolveCommand(BaseResolveTestCase): await self.support_abandon(claim_id1) await self.assertResolvesToClaimId('@foo', claim_id2) + async def test_resolve_duplicate_name_in_channel(self): + db_resolve = self.conductor.spv_node.server.db.resolve + # first one remains winner unless something else changes + channel_id = self.get_claim_id(await self.channel_create('@foo')) + + file_path = self.create_upload_file(data=b'hi!') + tx = await self.daemon.jsonrpc_stream_create('duplicate', '0.1', file_path=file_path, allow_duplicate_name=True, channel_id=channel_id) + await self.ledger.wait(tx) + + first_claim = tx.outputs[0].claim_id + + file_path = self.create_upload_file(data=b'hi!') + tx = await self.daemon.jsonrpc_stream_create('duplicate', '0.1', file_path=file_path, allow_duplicate_name=True, channel_id=channel_id) + await self.ledger.wait(tx) + duplicate_claim = tx.outputs[0].claim_id + await self.generate(1) + + stream, channel, _, _ = await db_resolve(f"@foo:{channel_id}/duplicate:{first_claim}") + self.assertEqual(stream.claim_hash.hex(), first_claim) + self.assertEqual(channel.claim_hash.hex(), channel_id) + stream, channel, _, _ = await db_resolve(f"@foo:{channel_id}/duplicate:{duplicate_claim}") + self.assertEqual(stream.claim_hash.hex(), duplicate_claim) + self.assertEqual(channel.claim_hash.hex(), channel_id) + async def test_advanced_resolve(self): claim_id1 = self.get_claim_id( await self.stream_create('foo', '0.7', allow_duplicate_name=True))