9301689383

Committed 30 May 2024 11:21AM UTC coverage: 56.987% (-0.1%) from 57.083%

Build # 9301689383

Build Type

Pull #3739

github

Committed by

web-flow

Commit Message

Merge 191627e4f into eb5da9511

Pull Request Pull Request #3739: Fix checking protocol version

Run Details

2804 of 6558 branches covered (42.76%)

Branch coverage included in aggregate %.

4 of 8 new or added lines in 1 file covered. (50.0%)

17 existing lines in 4 files now uncovered.

12281 of 19913 relevant lines covered (61.67%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

21.43

/lbry/wallet/network.py

import logging
import asyncio
import json
import socket
import random
from time import perf_counter
from collections import defaultdict
from typing import Dict, Optional, Tuple
import aiohttp

from lbry import __version__
from lbry.utils import resolve_host
from lbry.error import IncompatibleWalletServerError
from lbry.wallet.rpc import RPCSession as BaseClientSession, Connector, RPCError, ProtocolError
from lbry.wallet.stream import StreamController
from lbry.wallet.udp import SPVStatusClientProtocol, SPVPong
from lbry.conf import KnownHubsList

log = logging.getLogger(__name__)


class ClientSession(BaseClientSession):
    def __init__(self, *args, network: 'Network', server, timeout=30, concurrency=32, **kwargs):
        self.network = network
        self.server = server
        super().__init__(*args, **kwargs)
        self.framer.max_size = self.max_errors = 1 << 32
        self.timeout = timeout
        self.max_seconds_idle = timeout * 2
        self.response_time: Optional[float] = None
        self.connection_latency: Optional[float] = None
        self._response_samples = 0
        self._concurrency = asyncio.Semaphore(concurrency)

    @property
    def concurrency(self):
        return self._concurrency._value

    @property
    def available(self):
        return not self.is_closing() and self.response_time is not None

    @property
    def server_address_and_port(self) -> Optional[Tuple[str, int]]:
        if not self.transport:
            return None
        return self.transport.get_extra_info('peername')

    async def send_timed_server_version_request(self, args=(), timeout=None):
        timeout = timeout or self.timeout
        log.debug("send version request to %s:%i", *self.server)
        start = perf_counter()
        result = await asyncio.wait_for(
            super().send_request('server.version', args), timeout=timeout
        )
        current_response_time = perf_counter() - start
        response_sum = (self.response_time or 0) * self._response_samples + current_response_time
        self.response_time = response_sum / (self._response_samples + 1)
        self._response_samples += 1
        return result

    async def send_request(self, method, args=()):
        log.debug("send %s%s to %s:%i (%i timeout)", method, tuple(args), self.server[0], self.server[1], self.timeout)
        try:
            await self._concurrency.acquire()
            if method == 'server.version':
                return await self.send_timed_server_version_request(args, self.timeout)
            request = asyncio.ensure_future(super().send_request(method, args))
            while not request.done():
                done, pending = await asyncio.wait([request], timeout=self.timeout)
                if pending:
                    log.debug("Time since last packet: %s", perf_counter() - self.last_packet_received)
                    if (perf_counter() - self.last_packet_received) < self.timeout:
                        continue
                    log.warning("timeout sending %s to %s:%i", method, *self.server)
                    raise asyncio.TimeoutError
                if done:
                    try:
                        return request.result()
                    except ConnectionResetError:
                        log.error(
                            "wallet server (%s) reset connection upon our %s request, json of %i args is %i bytes",
                            self.server[0], method, len(args), len(json.dumps(args))
                        )
                        raise
        except (RPCError, ProtocolError) as e:
            log.warning("Wallet server (%s:%i) returned an error. Code: %s Message: %s",
                        *self.server, *e.args)
            raise e
        except ConnectionError:
            log.warning("connection to %s:%i lost", *self.server)
            self.synchronous_close()
            raise
        except asyncio.CancelledError:
            log.warning("cancelled sending %s to %s:%i", method, *self.server)
            # self.synchronous_close()
            raise
        finally:
            self._concurrency.release()

    async def ensure_server_version(self, required=None, timeout=3):
        required = required or self.network.PROTOCOL_MAX_VERSION
        response = await asyncio.wait_for(
            self.send_request('server.version', [self.network.CLIENT_NAME, required]), timeout=timeout
        )
        if tuple(int(piece) for piece in response[1].split(".")) >= self.network.PROTOCOL_MIN_VERSION:
            return response
        raise IncompatibleWalletServerError(*self.server)

    async def keepalive_loop(self, timeout=3, max_idle=60):
        try:
            while True:
                now = perf_counter()
                if min(self.last_send, self.last_packet_received) + max_idle < now:
                    await asyncio.wait_for(
                        self.send_request('server.ping', []), timeout=timeout
                    )
                else:
                    await asyncio.sleep(max(0, max_idle - (now - self.last_send)))
        except (Exception, asyncio.CancelledError) as err:
            if isinstance(err, asyncio.CancelledError):
                log.info("closing connection to %s:%i", *self.server)
            else:
                log.exception("lost connection to spv")
        finally:
            if not self.is_closing():
                self._close()

    async def create_connection(self, timeout=6):
        connector = Connector(lambda: self, *self.server)
        start = perf_counter()
        await asyncio.wait_for(connector.create_connection(), timeout=timeout)
        self.connection_latency = perf_counter() - start

    async def handle_request(self, request):
        controller = self.network.subscription_controllers[request.method]
        controller.add(request.args)

    def connection_lost(self, exc):
        log.debug("Connection lost: %s:%d", *self.server)
        super().connection_lost(exc)
        self.response_time = None
        self.connection_latency = None
        self._response_samples = 0
        # self._on_disconnect_controller.add(True)
        if self.network:
            self.network.disconnect()


class Network:

    CLIENT_VERSION = __version__
    CLIENT_NAME = "LBRY SDK " + CLIENT_VERSION

    PROTOCOL_MIN_VERSION = (0, 65, 0)
    PROTOCOL_MAX_VERSION = __version__

    def __init__(self, ledger):
        self.ledger = ledger
        self.client: Optional[ClientSession] = None
        self.server_features = None
        # self._switch_task: Optional[asyncio.Task] = None
        self.running = False
        self.remote_height: int = 0

        self._on_connected_controller = StreamController()
        self.on_connected = self._on_connected_controller.stream

        self._on_header_controller = StreamController(merge_repeated_events=True)
        self.on_header = self._on_header_controller.stream

        self._on_status_controller = StreamController(merge_repeated_events=True)
        self.on_status = self._on_status_controller.stream

        self._on_hub_controller = StreamController(merge_repeated_events=True)
        self.on_hub = self._on_hub_controller.stream

        self.subscription_controllers = {
            'blockchain.headers.subscribe': self._on_header_controller,
            'blockchain.address.subscribe': self._on_status_controller,
            'blockchain.peers.subscribe': self._on_hub_controller,
        }

        self.aiohttp_session: Optional[aiohttp.ClientSession] = None
        self._urgent_need_reconnect = asyncio.Event()
        self._loop_task: Optional[asyncio.Task] = None
        self._keepalive_task: Optional[asyncio.Task] = None

    @property
    def config(self):
        return self.ledger.config

    @property
    def known_hubs(self):
        if 'known_hubs' not in self.config:
            return KnownHubsList()
        return self.config['known_hubs']

    @property
    def jurisdiction(self):
        return self.config.get("jurisdiction")

    def disconnect(self):
        if self._keepalive_task and not self._keepalive_task.done():
            self._keepalive_task.cancel()
        self._keepalive_task = None

    async def start(self):
        if not self.running:
            self.running = True
            self.aiohttp_session = aiohttp.ClientSession()
            self.on_header.listen(self._update_remote_height)
            self.on_hub.listen(self._update_hubs)
            self._loop_task = asyncio.create_task(self.network_loop())
            self._urgent_need_reconnect.set()

        def loop_task_done_callback(f):
            try:
                f.result()
            except (Exception, asyncio.CancelledError):
                if self.running:
                    log.exception("wallet server connection loop crashed")

        self._loop_task.add_done_callback(loop_task_done_callback)

    async def resolve_spv_dns(self):
        hostname_to_ip = {}
        ip_to_hostnames = defaultdict(list)

        async def resolve_spv(server, port):
            try:
                server_addr = await resolve_host(server, port, 'udp')
                hostname_to_ip[server] = (server_addr, port)
                ip_to_hostnames[(server_addr, port)].append(server)
            except socket.error:
                log.warning("error looking up dns for spv server %s:%i", server, port)
            except Exception:
                log.exception("error looking up dns for spv server %s:%i", server, port)

        # accumulate the dns results
        if self.config.get('explicit_servers', []):
            hubs = self.config['explicit_servers']
        elif self.known_hubs:
            hubs = self.known_hubs
        else:
            hubs = self.config['default_servers']
        await asyncio.gather(*(resolve_spv(server, port) for (server, port) in hubs))
        return hostname_to_ip, ip_to_hostnames

    async def get_n_fastest_spvs(self, timeout=3.0) -> Dict[Tuple[str, int], Optional[SPVPong]]:
        loop = asyncio.get_event_loop()
        pong_responses = asyncio.Queue()
        connection = SPVStatusClientProtocol(pong_responses)
        sent_ping_timestamps = {}
        _, ip_to_hostnames = await self.resolve_spv_dns()
        n = len(ip_to_hostnames)
        log.info("%i possible spv servers to try (%i urls in config)", n, len(self.config.get('explicit_servers', [])))
        pongs = {}
        known_hubs = self.known_hubs
        try:
            await loop.create_datagram_endpoint(lambda: connection, ('0.0.0.0', 0))
            # could raise OSError if it cant bind
            start = perf_counter()
            for server in ip_to_hostnames:
                connection.ping(server)
                sent_ping_timestamps[server] = perf_counter()
            while len(pongs) < n:
                (remote, ts), pong = await asyncio.wait_for(pong_responses.get(), timeout - (perf_counter() - start))
                latency = ts - start
                log.info("%s:%i has latency of %sms (available: %s, height: %i)",
                         '/'.join(ip_to_hostnames[remote]), remote[1], round(latency * 1000, 2),
                         pong.available, pong.height)

                known_hubs.hubs.setdefault((ip_to_hostnames[remote][0], remote[1]), {}).update(
                    {"country": pong.country_name}
                )
                if pong.available:
                    pongs[(ip_to_hostnames[remote][0], remote[1])] = pong
            return pongs
        except asyncio.TimeoutError:
            if pongs:
                log.info("%i/%i probed spv servers are accepting connections", len(pongs), len(ip_to_hostnames))
                return pongs
            else:
                log.warning("%i spv status probes failed, retrying later. servers tried: %s",
                            len(sent_ping_timestamps),
                            ', '.join('/'.join(hosts) + f' ({ip})' for ip, hosts in ip_to_hostnames.items()))
                random_server = random.choice(list(ip_to_hostnames.keys()))
                host, port = random_server
                log.warning("trying fallback to randomly selected spv: %s:%i", host, port)
                known_hubs.hubs.setdefault((host, port), {})
                return {(host, port): None}
        finally:
            connection.close()

    async def connect_to_fastest(self) -> Optional[ClientSession]:
        fastest_spvs = await self.get_n_fastest_spvs()
        for (host, port), pong in fastest_spvs.items():
            if (pong is not None and self.jurisdiction is not None) and \
                    (pong.country_name != self.jurisdiction):
                continue
            client = ClientSession(network=self, server=(host, port), timeout=self.config.get('hub_timeout', 30),
                                   concurrency=self.config.get('concurrent_hub_requests', 30))
            try:
                await client.create_connection()
                log.info("Connected to spv server %s:%i", host, port)
                await client.ensure_server_version()
                return client
            except (asyncio.TimeoutError, ConnectionError, OSError, IncompatibleWalletServerError, RPCError):
                log.warning("Connecting to %s:%d failed", host, port)
                client._close()
        return

    async def network_loop(self):
        sleep_delay = 30
        while self.running:
            await asyncio.wait(
                map(asyncio.create_task, [asyncio.sleep(30), self._urgent_need_reconnect.wait()]),
                return_when=asyncio.FIRST_COMPLETED
            )
            if self._urgent_need_reconnect.is_set():
                sleep_delay = 30
            self._urgent_need_reconnect.clear()
            if not self.is_connected:
                client = await self.connect_to_fastest()
                if not client:
                    log.warning("failed to connect to any spv servers, retrying later")
                    sleep_delay *= 2
                    sleep_delay = min(sleep_delay, 300)
                    continue
                log.debug("get spv server features %s:%i", *client.server)
                features = await client.send_request('server.features', [])
                self.client, self.server_features = client, features
                log.debug("discover other hubs %s:%i", *client.server)
                await self._update_hubs(await client.send_request('server.peers.subscribe', []))
                log.info("subscribe to headers %s:%i", *client.server)
                self._update_remote_height((await self.subscribe_headers(),))
                self._on_connected_controller.add(True)
                server_str = "%s:%i" % client.server
                log.info("maintaining connection to spv server %s", server_str)
                self._keepalive_task = asyncio.create_task(self.client.keepalive_loop())
                try:
                    if not self._urgent_need_reconnect.is_set():
                        await asyncio.wait(
                            [self._keepalive_task, asyncio.create_task(self._urgent_need_reconnect.wait())],
                            return_when=asyncio.FIRST_COMPLETED
                        )
                    else:
                        await self._keepalive_task
                    if self._urgent_need_reconnect.is_set():
                        log.warning("urgent reconnect needed")
                    if self._keepalive_task and not self._keepalive_task.done():
                        self._keepalive_task.cancel()
                except asyncio.CancelledError:
                    pass
                finally:
                    self._keepalive_task = None
                    self.client = None
                    self.server_features = None
                    log.info("connection lost to %s", server_str)
        log.info("network loop finished")

    async def stop(self):
        self.running = False
        self.disconnect()
        if self._loop_task and not self._loop_task.done():
            self._loop_task.cancel()
        self._loop_task = None
        if self.aiohttp_session:
            await self.aiohttp_session.close()
            self.aiohttp_session = None

    @property
    def is_connected(self):
        return self.client and not self.client.is_closing()

    def rpc(self, list_or_method, args, restricted=True, session: Optional[ClientSession] = None):
        if session or self.is_connected:
            session = session or self.client
            return session.send_request(list_or_method, args)
        else:
            self._urgent_need_reconnect.set()
            raise ConnectionError("Attempting to send rpc request when connection is not available.")

    async def retriable_call(self, function, *args, **kwargs):
        while self.running:
            if not self.is_connected:
                log.warning("Wallet server unavailable, waiting for it to come back and retry.")
                self._urgent_need_reconnect.set()
                await self.on_connected.first
            try:
                return await function(*args, **kwargs)
            except asyncio.TimeoutError:
                log.warning("Wallet server call timed out, retrying.")
            except ConnectionError:
                log.warning("connection error")
        raise asyncio.CancelledError()  # if we got here, we are shutting down

    def _update_remote_height(self, header_args):
        self.remote_height = header_args[0]["height"]

    async def _update_hubs(self, hubs):
        if hubs and hubs != ['']:
            try:
                if self.known_hubs.add_hubs(hubs):
                    self.known_hubs.save()
            except Exception:
                log.exception("could not add hubs: %s", hubs)

    def get_transaction(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.get', [tx_hash], restricted)

    def get_transaction_batch(self, txids, restricted=True):
        # use any server if its old, otherwise restrict to who gave us the history
        return self.rpc('blockchain.transaction.get_batch', txids, 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)

    def get_merkle(self, tx_hash, height):
        restricted = 0 > height > self.remote_height - 10
        return self.rpc('blockchain.transaction.get_merkle', [tx_hash, height], restricted)

    def get_headers(self, height, count=10000, b64=False):
        restricted = height >= self.remote_height - 100
        return self.rpc('blockchain.block.headers', [height, count, 0, b64], restricted)

    #  --- Subscribes, history and broadcasts are always aimed towards the master client directly
    def get_history(self, address):
        return self.rpc('blockchain.address.get_history', [address], True)

    def broadcast(self, raw_transaction):
        return self.rpc('blockchain.transaction.broadcast', [raw_transaction], True)

    def subscribe_headers(self):
        return self.rpc('blockchain.headers.subscribe', [True], True)

    async def subscribe_address(self, address, *addresses):
        addresses = list((address, ) + addresses)
        server_addr_and_port = self.client.server_address_and_port  # on disconnect client will be None
        try:
            return await self.rpc('blockchain.address.subscribe', addresses, True)
        except asyncio.TimeoutError:
            log.warning(
                "timed out subscribing to addresses from %s:%i",
                *server_addr_and_port
            )
            # abort and cancel, we can't lose a subscription, it will happen again on reconnect
            if self.client:
                self.client.abort()
            raise asyncio.CancelledError()

    def unsubscribe_address(self, address):
        return self.rpc('blockchain.address.unsubscribe', [address], True)

    def get_server_features(self):
        return self.rpc('server.features', (), restricted=True)

    # def get_claims_by_ids(self, claim_ids):
    #     return self.rpc('blockchain.claimtrie.getclaimsbyids', claim_ids)

    def get_claim_by_id(self, claim_id):
        return self.rpc('blockchain.claimtrie.getclaimbyid', [claim_id])

    def resolve(self, urls, session_override=None):
        return self.rpc('blockchain.claimtrie.resolve', urls, False, session_override)

    def claim_search(self, session_override=None, **kwargs):
        return self.rpc('blockchain.claimtrie.search', kwargs, False, session_override)

    async def sum_supports(self, server, **kwargs):
        message = {"method": "support_sum", "params": kwargs}
        async with self.aiohttp_session.post(server, json=message) as r:
            result = await r.json()
            return result['result']

1	import logging	3✔
2	import asyncio	3✔
3	import json	3✔
4	import socket	3✔
5	import random	3✔
6	from time import perf_counter	3✔
7	from collections import defaultdict	3✔
8	from typing import Dict, Optional, Tuple	3✔
9	import aiohttp	3✔
10
11	from lbry import __version__	3✔
12	from lbry.utils import resolve_host	3✔
13	from lbry.error import IncompatibleWalletServerError	3✔
14	from lbry.wallet.rpc import RPCSession as BaseClientSession, Connector, RPCError, ProtocolError	3✔
15	from lbry.wallet.stream import StreamController	3✔
16	from lbry.wallet.udp import SPVStatusClientProtocol, SPVPong	3✔
17	from lbry.conf import KnownHubsList	3✔
18
19	log = logging.getLogger(__name__)	3✔
20
21
22	class ClientSession(BaseClientSession):	3✔
23	def __init__(self, args, network: 'Network', server, timeout=30, concurrency=32, *kwargs):	3✔
24	self.network = network	1✔
25	self.server = server	1✔
26	super().__init__(args, *kwargs)	1✔
27	self.framer.max_size = self.max_errors = 1 << 32	1✔
28	self.timeout = timeout	1✔
29	self.max_seconds_idle = timeout * 2	1✔
30	self.response_time: Optional[float] = None	1✔
31	self.connection_latency: Optional[float] = None	1✔
32	self._response_samples = 0	1✔
33	self._concurrency = asyncio.Semaphore(concurrency)	1✔
34
35	@property	3✔
36	def concurrency(self):	3✔
37	return self._concurrency._value	×
38
39	@property	3✔
40	def available(self):	3✔
41	return not self.is_closing() and self.response_time is not None	×
42
43	@property	3✔
44	def server_address_and_port(self) -> Optional[Tuple[str, int]]:	3✔
45	if not self.transport:	×
46	return None	×
47	return self.transport.get_extra_info('peername')	×
48
49	async def send_timed_server_version_request(self, args=(), timeout=None):	3✔
50	timeout = timeout or self.timeout	×
51	log.debug("send version request to %s:%i", *self.server)	×
52	start = perf_counter()	×
53	result = await asyncio.wait_for(	×
54	super().send_request('server.version', args), timeout=timeout
55	)
56	current_response_time = perf_counter() - start	×
57	response_sum = (self.response_time or 0) * self._response_samples + current_response_time	×
58	self.response_time = response_sum / (self._response_samples + 1)	×
59	self._response_samples += 1	×
60	return result	×
61
62	async def send_request(self, method, args=()):	3✔
63	log.debug("send %s%s to %s:%i (%i timeout)", method, tuple(args), self.server[0], self.server[1], self.timeout)	×
64	try:	×
65	await self._concurrency.acquire()	×
66	if method == 'server.version':	×
67	return await self.send_timed_server_version_request(args, self.timeout)	×
68	request = asyncio.ensure_future(super().send_request(method, args))	×
69	while not request.done():	×
70	done, pending = await asyncio.wait([request], timeout=self.timeout)	×
71	if pending:	×
72	log.debug("Time since last packet: %s", perf_counter() - self.last_packet_received)	×
73	if (perf_counter() - self.last_packet_received) < self.timeout:	×
74	continue	×
75	log.warning("timeout sending %s to %s:%i", method, *self.server)	×
76	raise asyncio.TimeoutError	×
77	if done:	×
78	try:	×
79	return request.result()	×
80	except ConnectionResetError:	×
81	log.error(	×
82	"wallet server (%s) reset connection upon our %s request, json of %i args is %i bytes",
83	self.server[0], method, len(args), len(json.dumps(args))
84	)
85	raise	×
86	except (RPCError, ProtocolError) as e:	×
87	log.warning("Wallet server (%s:%i) returned an error. Code: %s Message: %s",	×
88	self.server, e.args)
89	raise e	×
90	except ConnectionError:	×
91	log.warning("connection to %s:%i lost", *self.server)	×
92	self.synchronous_close()	×
93	raise	×
94	except asyncio.CancelledError:	×
95	log.warning("cancelled sending %s to %s:%i", method, *self.server)	×
96	# self.synchronous_close()
97	raise	×
98	finally:
99	self._concurrency.release()	×
100
101	async def ensure_server_version(self, required=None, timeout=3):	3✔
NEW 102	required = required or self.network.PROTOCOL_MAX_VERSION	×
103	response = await asyncio.wait_for(	×
104	self.send_request('server.version', [self.network.CLIENT_NAME, required]), timeout=timeout
105	)
NEW 106	if tuple(int(piece) for piece in response[1].split(".")) >= self.network.PROTOCOL_MIN_VERSION:	×
NEW 107	return response	×
NEW 108	raise IncompatibleWalletServerError(*self.server)	×
109
110	async def keepalive_loop(self, timeout=3, max_idle=60):	3✔
111	try:	×
112	while True:
113	now = perf_counter()	×
114	if min(self.last_send, self.last_packet_received) + max_idle < now:	×
115	await asyncio.wait_for(	×
116	self.send_request('server.ping', []), timeout=timeout
117	)
118	else:
119	await asyncio.sleep(max(0, max_idle - (now - self.last_send)))	×
120	except (Exception, asyncio.CancelledError) as err:	×
121	if isinstance(err, asyncio.CancelledError):	×
122	log.info("closing connection to %s:%i", *self.server)	×
123	else:
124	log.exception("lost connection to spv")	×
125	finally:
126	if not self.is_closing():	×
127	self._close()	×
128
129	async def create_connection(self, timeout=6):	3✔
130	connector = Connector(lambda: self, *self.server)	×
131	start = perf_counter()	×
132	await asyncio.wait_for(connector.create_connection(), timeout=timeout)	×
133	self.connection_latency = perf_counter() - start	×
134
135	async def handle_request(self, request):	3✔
136	controller = self.network.subscription_controllers[request.method]	×
137	controller.add(request.args)	×
138
139	def connection_lost(self, exc):	3✔
140	log.debug("Connection lost: %s:%d", *self.server)	×
141	super().connection_lost(exc)	×
142	self.response_time = None	×
143	self.connection_latency = None	×
144	self._response_samples = 0	×
145	# self._on_disconnect_controller.add(True)
146	if self.network:	×
147	self.network.disconnect()	×
148
149
150	class Network:	3✔
151
152	CLIENT_VERSION = __version__	3✔
153	CLIENT_NAME = "LBRY SDK " + CLIENT_VERSION	3✔
154
155	PROTOCOL_MIN_VERSION = (0, 65, 0)	3✔
156	PROTOCOL_MAX_VERSION = __version__	3✔
157
158	def __init__(self, ledger):	3✔
159	self.ledger = ledger	1✔
160	self.client: Optional[ClientSession] = None	1✔
161	self.server_features = None	1✔
162	# self._switch_task: Optional[asyncio.Task] = None
163	self.running = False	1✔
164	self.remote_height: int = 0	1✔
165
166	self._on_connected_controller = StreamController()	1✔
167	self.on_connected = self._on_connected_controller.stream	1✔
168
169	self._on_header_controller = StreamController(merge_repeated_events=True)	1✔
170	self.on_header = self._on_header_controller.stream	1✔
171
172	self._on_status_controller = StreamController(merge_repeated_events=True)	1✔
173	self.on_status = self._on_status_controller.stream	1✔
174
175	self._on_hub_controller = StreamController(merge_repeated_events=True)	1✔
176	self.on_hub = self._on_hub_controller.stream	1✔
177
178	self.subscription_controllers = {	1✔
179	'blockchain.headers.subscribe': self._on_header_controller,
180	'blockchain.address.subscribe': self._on_status_controller,
181	'blockchain.peers.subscribe': self._on_hub_controller,
182	}
183
184	self.aiohttp_session: Optional[aiohttp.ClientSession] = None	1✔
185	self._urgent_need_reconnect = asyncio.Event()	1✔
186	self._loop_task: Optional[asyncio.Task] = None	1✔
187	self._keepalive_task: Optional[asyncio.Task] = None	1✔
188
189	@property	3✔
190	def config(self):	3✔
191	return self.ledger.config	×
192
193	@property	3✔
194	def known_hubs(self):	3✔
195	if 'known_hubs' not in self.config:	×
196	return KnownHubsList()	×
197	return self.config['known_hubs']	×
198
199	@property	3✔
200	def jurisdiction(self):	3✔
201	return self.config.get("jurisdiction")	×
202
203	def disconnect(self):	3✔
204	if self._keepalive_task and not self._keepalive_task.done():	×
205	self._keepalive_task.cancel()	×
206	self._keepalive_task = None	×
207
208	async def start(self):	3✔
209	if not self.running:	×
210	self.running = True	×
211	self.aiohttp_session = aiohttp.ClientSession()	×
212	self.on_header.listen(self._update_remote_height)	×
213	self.on_hub.listen(self._update_hubs)	×
214	self._loop_task = asyncio.create_task(self.network_loop())	×
215	self._urgent_need_reconnect.set()	×
216
217	def loop_task_done_callback(f):	×
218	try:	×
219	f.result()	×
220	except (Exception, asyncio.CancelledError):	×
221	if self.running:	×
222	log.exception("wallet server connection loop crashed")	×
223
224	self._loop_task.add_done_callback(loop_task_done_callback)	×
225
226	async def resolve_spv_dns(self):	3✔
227	hostname_to_ip = {}	×
228	ip_to_hostnames = defaultdict(list)	×
229
230	async def resolve_spv(server, port):	×
231	try:	×
232	server_addr = await resolve_host(server, port, 'udp')	×
233	hostname_to_ip[server] = (server_addr, port)	×
234	ip_to_hostnames[(server_addr, port)].append(server)	×
235	except socket.error:	×
236	log.warning("error looking up dns for spv server %s:%i", server, port)	×
237	except Exception:	×
238	log.exception("error looking up dns for spv server %s:%i", server, port)	×
239
240	# accumulate the dns results
241	if self.config.get('explicit_servers', []):	×
242	hubs = self.config['explicit_servers']	×
243	elif self.known_hubs:	×
244	hubs = self.known_hubs	×
245	else:
246	hubs = self.config['default_servers']	×
247	await asyncio.gather(*(resolve_spv(server, port) for (server, port) in hubs))	×
248	return hostname_to_ip, ip_to_hostnames	×
249
250	async def get_n_fastest_spvs(self, timeout=3.0) -> Dict[Tuple[str, int], Optional[SPVPong]]:	3✔
251	loop = asyncio.get_event_loop()	×
252	pong_responses = asyncio.Queue()	×
253	connection = SPVStatusClientProtocol(pong_responses)	×
254	sent_ping_timestamps = {}	×
255	_, ip_to_hostnames = await self.resolve_spv_dns()	×
256	n = len(ip_to_hostnames)	×
257	log.info("%i possible spv servers to try (%i urls in config)", n, len(self.config.get('explicit_servers', [])))	×
258	pongs = {}	×
259	known_hubs = self.known_hubs	×
260	try:	×
261	await loop.create_datagram_endpoint(lambda: connection, ('0.0.0.0', 0))	×
262	# could raise OSError if it cant bind
263	start = perf_counter()	×
264	for server in ip_to_hostnames:	×
265	connection.ping(server)	×
266	sent_ping_timestamps[server] = perf_counter()	×
267	while len(pongs) < n:	×
268	(remote, ts), pong = await asyncio.wait_for(pong_responses.get(), timeout - (perf_counter() - start))	×
269	latency = ts - start	×
270	log.info("%s:%i has latency of %sms (available: %s, height: %i)",	×
271	'/'.join(ip_to_hostnames[remote]), remote[1], round(latency * 1000, 2),
272	pong.available, pong.height)
273
274	known_hubs.hubs.setdefault((ip_to_hostnames[remote][0], remote[1]), {}).update(	×
275	{"country": pong.country_name}
276	)
277	if pong.available:	×
278	pongs[(ip_to_hostnames[remote][0], remote[1])] = pong	×
279	return pongs	×
280	except asyncio.TimeoutError:	×
281	if pongs:	×
282	log.info("%i/%i probed spv servers are accepting connections", len(pongs), len(ip_to_hostnames))	×
283	return pongs	×
284	else:
285	log.warning("%i spv status probes failed, retrying later. servers tried: %s",	×
286	len(sent_ping_timestamps),
287	', '.join('/'.join(hosts) + f' ({ip})' for ip, hosts in ip_to_hostnames.items()))
288	random_server = random.choice(list(ip_to_hostnames.keys()))	×
289	host, port = random_server	×
290	log.warning("trying fallback to randomly selected spv: %s:%i", host, port)	×
291	known_hubs.hubs.setdefault((host, port), {})	×
292	return {(host, port): None}	×
293	finally:
294	connection.close()	×
295
296	async def connect_to_fastest(self) -> Optional[ClientSession]:	3✔
297	fastest_spvs = await self.get_n_fastest_spvs()	×
298	for (host, port), pong in fastest_spvs.items():	×
299	if (pong is not None and self.jurisdiction is not None) and \	×
300	(pong.country_name != self.jurisdiction):
301	continue	×
302	client = ClientSession(network=self, server=(host, port), timeout=self.config.get('hub_timeout', 30),	×
303	concurrency=self.config.get('concurrent_hub_requests', 30))
304	try:	×
305	await client.create_connection()	×
306	log.info("Connected to spv server %s:%i", host, port)	×
307	await client.ensure_server_version()	×
308	return client	×
309	except (asyncio.TimeoutError, ConnectionError, OSError, IncompatibleWalletServerError, RPCError):	×
310	log.warning("Connecting to %s:%d failed", host, port)	×
311	client._close()	×
312	return	×
313
314	async def network_loop(self):	3✔
315	sleep_delay = 30	×
316	while self.running:	×
317	await asyncio.wait(	×
318	map(asyncio.create_task, [asyncio.sleep(30), self._urgent_need_reconnect.wait()]),
319	return_when=asyncio.FIRST_COMPLETED
320	)
321	if self._urgent_need_reconnect.is_set():	×
322	sleep_delay = 30	×
323	self._urgent_need_reconnect.clear()	×
324	if not self.is_connected:	×
325	client = await self.connect_to_fastest()	×
326	if not client:	×
327	log.warning("failed to connect to any spv servers, retrying later")	×
328	sleep_delay *= 2	×
329	sleep_delay = min(sleep_delay, 300)	×
330	continue	×
331	log.debug("get spv server features %s:%i", *client.server)	×
332	features = await client.send_request('server.features', [])	×
333	self.client, self.server_features = client, features	×
334	log.debug("discover other hubs %s:%i", *client.server)	×
335	await self._update_hubs(await client.send_request('server.peers.subscribe', []))	×
336	log.info("subscribe to headers %s:%i", *client.server)	×
337	self._update_remote_height((await self.subscribe_headers(),))	×
338	self._on_connected_controller.add(True)	×
339	server_str = "%s:%i" % client.server	×
340	log.info("maintaining connection to spv server %s", server_str)	×
341	self._keepalive_task = asyncio.create_task(self.client.keepalive_loop())	×
342	try:	×
343	if not self._urgent_need_reconnect.is_set():	×
344	await asyncio.wait(	×
345	[self._keepalive_task, asyncio.create_task(self._urgent_need_reconnect.wait())],
346	return_when=asyncio.FIRST_COMPLETED
347	)
348	else:
349	await self._keepalive_task	×
350	if self._urgent_need_reconnect.is_set():	×
351	log.warning("urgent reconnect needed")	×
352	if self._keepalive_task and not self._keepalive_task.done():	×
353	self._keepalive_task.cancel()	×
354	except asyncio.CancelledError:	×
355	pass	×
356	finally:
357	self._keepalive_task = None	×
358	self.client = None	×
359	self.server_features = None	×
360	log.info("connection lost to %s", server_str)	×
361	log.info("network loop finished")	×
362
363	async def stop(self):	3✔
364	self.running = False	×
365	self.disconnect()	×
366	if self._loop_task and not self._loop_task.done():	×
367	self._loop_task.cancel()	×
368	self._loop_task = None	×
369	if self.aiohttp_session:	×
370	await self.aiohttp_session.close()	×
371	self.aiohttp_session = None	×
372
373	@property	3✔
374	def is_connected(self):	3✔
375	return self.client and not self.client.is_closing()	1✔
376
377	def rpc(self, list_or_method, args, restricted=True, session: Optional[ClientSession] = None):	3✔
378	if session or self.is_connected:	×
379	session = session or self.client	×
380	return session.send_request(list_or_method, args)	×
381	else:
382	self._urgent_need_reconnect.set()	×
383	raise ConnectionError("Attempting to send rpc request when connection is not available.")	×
384
385	async def retriable_call(self, function, args, *kwargs):	3✔
386	while self.running:	×
387	if not self.is_connected:	×
388	log.warning("Wallet server unavailable, waiting for it to come back and retry.")	×
389	self._urgent_need_reconnect.set()	×
390	await self.on_connected.first	×
391	try:	×
392	return await function(args, *kwargs)	×
393	except asyncio.TimeoutError:	×
394	log.warning("Wallet server call timed out, retrying.")	×
395	except ConnectionError:	×
396	log.warning("connection error")	×
397	raise asyncio.CancelledError() # if we got here, we are shutting down	×
398
399	def _update_remote_height(self, header_args):	3✔
400	self.remote_height = header_args[0]["height"]	×
401
402	async def _update_hubs(self, hubs):	3✔
403	if hubs and hubs != ['']:	×
404	try:	×
405	if self.known_hubs.add_hubs(hubs):	×
406	self.known_hubs.save()	×
407	except Exception:	×
408	log.exception("could not add hubs: %s", hubs)	×
409
410	def get_transaction(self, tx_hash, known_height=None):	3✔
411	# use any server if its old, otherwise restrict to who gave us the history
412	restricted = known_height in (None, -1, 0) or 0 > known_height > self.remote_height - 10	×
413	return self.rpc('blockchain.transaction.get', [tx_hash], restricted)	×
414
415	def get_transaction_batch(self, txids, restricted=True):	3✔
416	# use any server if its old, otherwise restrict to who gave us the history
417	return self.rpc('blockchain.transaction.get_batch', txids, restricted)	×
418
419	def get_transaction_and_merkle(self, tx_hash, known_height=None):	3✔
420	# use any server if its old, otherwise restrict to who gave us the history
421	restricted = known_height in (None, -1, 0) or 0 > known_height > self.remote_height - 10	×
422	return self.rpc('blockchain.transaction.info', [tx_hash], restricted)	×
423
424	def get_transaction_height(self, tx_hash, known_height=None):	3✔
425	restricted = not known_height or 0 > known_height > self.remote_height - 10	×
426	return self.rpc('blockchain.transaction.get_height', [tx_hash], restricted)	×
427
428	def get_merkle(self, tx_hash, height):	3✔
429	restricted = 0 > height > self.remote_height - 10	×
430	return self.rpc('blockchain.transaction.get_merkle', [tx_hash, height], restricted)	×
431
432	def get_headers(self, height, count=10000, b64=False):	3✔
433	restricted = height >= self.remote_height - 100	×
434	return self.rpc('blockchain.block.headers', [height, count, 0, b64], restricted)	×
435
436	# --- Subscribes, history and broadcasts are always aimed towards the master client directly
437	def get_history(self, address):	3✔
438	return self.rpc('blockchain.address.get_history', [address], True)	×
439
440	def broadcast(self, raw_transaction):	3✔
441	return self.rpc('blockchain.transaction.broadcast', [raw_transaction], True)	×
442
443	def subscribe_headers(self):	3✔
444	return self.rpc('blockchain.headers.subscribe', [True], True)	×
445
446	async def subscribe_address(self, address, *addresses):	3✔
447	addresses = list((address, ) + addresses)	×
448	server_addr_and_port = self.client.server_address_and_port # on disconnect client will be None	×
449	try:	×
450	return await self.rpc('blockchain.address.subscribe', addresses, True)	×
451	except asyncio.TimeoutError:	×
452	log.warning(	×
453	"timed out subscribing to addresses from %s:%i",
454	*server_addr_and_port
455	)
456	# abort and cancel, we can't lose a subscription, it will happen again on reconnect
457	if self.client:	×
458	self.client.abort()	×
459	raise asyncio.CancelledError()	×
460
461	def unsubscribe_address(self, address):	3✔
462	return self.rpc('blockchain.address.unsubscribe', [address], True)	×
463
464	def get_server_features(self):	3✔
465	return self.rpc('server.features', (), restricted=True)	×
466
467	# def get_claims_by_ids(self, claim_ids):
468	# return self.rpc('blockchain.claimtrie.getclaimsbyids', claim_ids)
469
470	def get_claim_by_id(self, claim_id):	3✔
471	return self.rpc('blockchain.claimtrie.getclaimbyid', [claim_id])	×
472
473	def resolve(self, urls, session_override=None):	3✔
474	return self.rpc('blockchain.claimtrie.resolve', urls, False, session_override)	×
475
476	def claim_search(self, session_override=None, **kwargs):	3✔
477	return self.rpc('blockchain.claimtrie.search', kwargs, False, session_override)	×
478
479	async def sum_supports(self, server, **kwargs):	3✔
480	message = {"method": "support_sum", "params": kwargs}	×
481	async with self.aiohttp_session.post(server, json=message) as r:	×
482	result = await r.json()	×
483	return result['result']	×

lbryio / lbry-sdk / 9301689383

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous