402 lines
16 KiB
Python
402 lines
16 KiB
Python
# Copyright (c) 2016-2017, Neil Booth
|
|
#
|
|
# All rights reserved.
|
|
#
|
|
# See the file "LICENCE" for information about the copyright
|
|
# and warranty status of this software.
|
|
|
|
'''Class for handling asynchronous connections to a blockchain
|
|
daemon.'''
|
|
|
|
import asyncio
|
|
import itertools
|
|
import time
|
|
from typing import TYPE_CHECKING, Type, Sequence, Any, Iterable, Optional
|
|
|
|
import aiohttp
|
|
from aiorpcx import JSONRPC
|
|
|
|
from electrumx.lib.util import (class_logger, hex_to_bytes, json_deserialize,
|
|
json_serialize)
|
|
|
|
if TYPE_CHECKING:
|
|
from electrumx.lib.coins import Coin
|
|
|
|
|
|
class DaemonError(Exception):
|
|
'''Raised when the daemon returns an error in its results.'''
|
|
|
|
|
|
class WarmingUpError(Exception):
|
|
'''Internal - when the daemon is warming up.'''
|
|
|
|
|
|
class ServiceRefusedError(Exception):
|
|
'''Internal - when the daemon doesn't provide a JSON response, only an HTTP error, for
|
|
some reason.'''
|
|
|
|
|
|
class DaemonParseError(Exception):
|
|
'''Internal - when the daemon returns an RPC_PARSE_ERROR.'''
|
|
|
|
|
|
class Daemon:
|
|
'''Handles connections to a daemon at the given URL.'''
|
|
|
|
RPC_IN_WARMUP = -28
|
|
RPC_PARSE_ERROR = -32700
|
|
|
|
id_counter = itertools.count()
|
|
|
|
def __init__(
|
|
self,
|
|
coin: Type['Coin'],
|
|
url,
|
|
*,
|
|
max_workqueue=10,
|
|
init_retry=0.25,
|
|
max_retry=4.0,
|
|
):
|
|
self.coin = coin
|
|
self.logger = class_logger(__name__, self.__class__.__name__)
|
|
self.url_index = None
|
|
self.urls = []
|
|
self.set_url(url)
|
|
# Limit concurrent RPC calls to this number.
|
|
# See DEFAULT_HTTP_WORKQUEUE in the daemon, which is typically 16
|
|
self.workqueue_semaphore = asyncio.Semaphore(value=max_workqueue)
|
|
self.init_retry = init_retry
|
|
self.max_retry = max_retry
|
|
self._height = None
|
|
self.height_changed = asyncio.Event()
|
|
self.available_rpcs = {}
|
|
self.session = None
|
|
|
|
self._networkinfo_cache = (None, 0)
|
|
self._networkinfo_lock = asyncio.Lock()
|
|
|
|
self._mempoolinfo_cache = (None, 0)
|
|
self._mempoolinfo_lock = asyncio.Lock()
|
|
|
|
async def __aenter__(self):
|
|
self.session = aiohttp.ClientSession(connector=self.connector())
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc_value, traceback):
|
|
await self.session.close()
|
|
self.session = None
|
|
|
|
def connector(self):
|
|
return None
|
|
|
|
async def check_daemon_version(self):
|
|
assert self.session is not None and self.coin is not None, f"{self.session=}, {self.coin=}"
|
|
if self.coin.MIN_REQUIRED_DAEMON_VERSION is None:
|
|
return
|
|
|
|
required_version = tuple(map(int, self.coin.MIN_REQUIRED_DAEMON_VERSION.split('.')))
|
|
network_info = await self.getnetworkinfo()
|
|
ni_version = network_info['version']
|
|
daemon_major = ni_version // 10_000
|
|
daemon_minor = (ni_version % 10_000) // 100
|
|
daemon_rev = ni_version % 100
|
|
daemon_version = (daemon_major, daemon_minor, daemon_rev)
|
|
if daemon_version < required_version:
|
|
raise RuntimeError(f"Daemon version {daemon_version=} < {required_version=}.")
|
|
|
|
async def check_daemon_indexes(self):
|
|
assert self.session is not None and self.coin is not None, f"{self.session=}, {self.coin=}"
|
|
if not self.coin.REQUIRED_DAEMON_INDEXES:
|
|
return
|
|
index_info = await self.getindexinfo()
|
|
for req_index in self.coin.REQUIRED_DAEMON_INDEXES:
|
|
if req_index not in index_info:
|
|
raise RuntimeError(
|
|
f"the daemon is missing required index: {req_index}. "
|
|
f"You should set {req_index}=1 in the daemon's config file, and reindex.")
|
|
if not index_info[req_index]["synced"]:
|
|
# Should we raise? Not raising allows syncing a fresh daemon and e-x "in parallel".
|
|
self.logger.warning(f"the daemon's required index {req_index!r} is still syncing!")
|
|
|
|
def set_url(self, url: str) -> None:
|
|
'''Set the URLS to the given list, and switch to the first one.'''
|
|
urls = url.split(',')
|
|
urls = [self.coin.sanitize_url(url) for url in urls]
|
|
for n, url in enumerate(urls):
|
|
status = '' if n else ' (current)'
|
|
logged_url = self.logged_url(url)
|
|
self.logger.info(f'daemon #{n + 1} at {logged_url}{status}')
|
|
self.url_index = 0
|
|
self.urls = urls
|
|
|
|
def current_url(self) -> str:
|
|
'''Returns the current daemon URL.'''
|
|
return self.urls[self.url_index]
|
|
|
|
def logged_url(self, url=None):
|
|
'''The host and port part, for logging.'''
|
|
url = url or self.current_url()
|
|
return url[url.rindex('@') + 1:]
|
|
|
|
def failover(self):
|
|
'''Call to fail-over to the next daemon URL.
|
|
|
|
Returns False if there is only one, otherwise True.
|
|
'''
|
|
if len(self.urls) > 1:
|
|
self.url_index = (self.url_index + 1) % len(self.urls)
|
|
self.logger.info(f'failing over to {self.logged_url()}')
|
|
return True
|
|
return False
|
|
|
|
async def _send_data(self, data):
|
|
async with self.workqueue_semaphore:
|
|
async with self.session.post(self.current_url(), data=data) as resp:
|
|
kind = resp.headers.get('Content-Type', None)
|
|
if kind == 'application/json':
|
|
return await resp.json(loads=json_deserialize)
|
|
text = await resp.text()
|
|
text = text.strip() or resp.reason
|
|
raise ServiceRefusedError(text)
|
|
|
|
async def _send(self, payload, processor):
|
|
'''Send a payload to be converted to JSON.
|
|
|
|
Handles temporary connection issues. Daemon response errors
|
|
are raise through DaemonError.
|
|
'''
|
|
def log_error(error):
|
|
nonlocal last_error_log, retry
|
|
now = time.monotonic()
|
|
if now - last_error_log > 60:
|
|
last_error_log = now
|
|
self.logger.error(f'{error}. Retrying occasionally...')
|
|
if retry == self.max_retry and self.failover():
|
|
retry = 0
|
|
|
|
on_good_message = None
|
|
last_error_log = 0
|
|
data = json_serialize(payload)
|
|
retry = self.init_retry
|
|
while True:
|
|
try:
|
|
result = await self._send_data(data)
|
|
result = processor(result)
|
|
if on_good_message:
|
|
self.logger.info(on_good_message)
|
|
return result
|
|
except asyncio.TimeoutError:
|
|
log_error('timeout error')
|
|
except aiohttp.ServerDisconnectedError:
|
|
log_error('disconnected')
|
|
on_good_message = 'connection restored'
|
|
except ConnectionResetError:
|
|
log_error('connection reset')
|
|
on_good_message = 'connection restored'
|
|
except aiohttp.ClientConnectionError:
|
|
log_error('connection problem - check your daemon is running')
|
|
on_good_message = 'connection restored'
|
|
except aiohttp.ClientError as e:
|
|
log_error(f'daemon error: {e}')
|
|
on_good_message = 'running normally'
|
|
except ServiceRefusedError as e:
|
|
log_error(f'daemon service refused: {e}')
|
|
on_good_message = 'running normally'
|
|
except WarmingUpError:
|
|
log_error('starting up checking blocks')
|
|
on_good_message = 'running normally'
|
|
except DaemonParseError as e:
|
|
log_error(f"received RPC_PARSE_ERROR ({e=!r}). request was: {data=!r}")
|
|
on_good_message = 'running normally'
|
|
|
|
await asyncio.sleep(retry)
|
|
retry = max(min(self.max_retry, retry * 2), self.init_retry)
|
|
|
|
async def _send_single(self, method, params=None):
|
|
'''Send a single request to the daemon.'''
|
|
def processor(result):
|
|
err = result['error']
|
|
if not err:
|
|
return result['result']
|
|
if err.get('code') == self.RPC_IN_WARMUP:
|
|
raise WarmingUpError
|
|
if err.get('code') == self.RPC_PARSE_ERROR:
|
|
raise DaemonParseError(err)
|
|
raise DaemonError(err)
|
|
|
|
payload = {'method': method, 'id': next(self.id_counter)}
|
|
if params:
|
|
payload['params'] = params
|
|
return await self._send(payload, processor)
|
|
|
|
async def _send_vector(self, method, params_iterable, replace_errs=False):
|
|
'''Send several requests of the same method.
|
|
|
|
The result will be an array of the same length as params_iterable.
|
|
If replace_errs is true, any item with an error is returned as None,
|
|
otherwise an exception is raised.'''
|
|
def processor(result):
|
|
errs = [item['error'] for item in result if item['error']]
|
|
if any(err.get('code') == self.RPC_IN_WARMUP for err in errs):
|
|
raise WarmingUpError
|
|
if any(err.get('code') == self.RPC_PARSE_ERROR for err in errs):
|
|
raise DaemonParseError
|
|
if not errs or replace_errs:
|
|
return [item['result'] for item in result]
|
|
raise DaemonError(errs)
|
|
|
|
payload = [{'method': method, 'params': p, 'id': next(self.id_counter)}
|
|
for p in params_iterable]
|
|
if payload:
|
|
return await self._send(payload, processor)
|
|
return []
|
|
|
|
async def _is_rpc_available(self, method: str) -> bool:
|
|
'''Return whether given RPC method is available in the daemon.
|
|
|
|
Results are cached and the daemon will generally not be queried with
|
|
the same method more than once.'''
|
|
available = self.available_rpcs.get(method)
|
|
if available is None:
|
|
available = True
|
|
try:
|
|
await self._send_single(method)
|
|
except DaemonError as e:
|
|
err = e.args[0]
|
|
error_code = err.get("code")
|
|
available = error_code != JSONRPC.METHOD_NOT_FOUND
|
|
self.available_rpcs[method] = available
|
|
return available
|
|
|
|
async def block_hex_hashes(self, first: int, count: int) -> Sequence[str]:
|
|
'''Return the hex hashes (hum) of count block starting at height first.'''
|
|
params_iterable = ((h, ) for h in range(first, first + count))
|
|
return await self._send_vector('getblockhash', params_iterable)
|
|
|
|
async def deserialised_block(self, bhash_hum: str) -> dict:
|
|
'''Return the deserialised block with the given hex hash.'''
|
|
return await self._send_single('getblock', (bhash_hum, True))
|
|
|
|
async def raw_blocks(self, bhashes_hum: Sequence[str]) -> Sequence[bytes]:
|
|
'''Return the raw binary blocks with the given hex hashes.'''
|
|
params_iterable = ((h, False) for h in bhashes_hum)
|
|
blocks = await self._send_vector('getblock', params_iterable)
|
|
# Convert hex string to bytes
|
|
return [hex_to_bytes(block) for block in blocks]
|
|
|
|
async def mempool_txids_hum(self) -> Sequence[str]:
|
|
'''Update our record of the daemon's mempool hashes.'''
|
|
return await self._send_single('getrawmempool')
|
|
|
|
async def estimatefee(self, block_count, estimate_mode=None):
|
|
'''Return the fee estimate for the block count. Units are whole
|
|
currency units per KB, e.g. 0.00000995, or -1 if no estimate
|
|
is available.
|
|
'''
|
|
if estimate_mode:
|
|
args = (block_count, estimate_mode)
|
|
else:
|
|
args = (block_count, )
|
|
if await self._is_rpc_available('estimatesmartfee'):
|
|
estimate = await self._send_single('estimatesmartfee', args)
|
|
return estimate.get('feerate', -1)
|
|
return await self._send_single('estimatefee', args)
|
|
|
|
async def getnetworkinfo(self):
|
|
'''Return the result of the 'getnetworkinfo' RPC call.'''
|
|
async with self._networkinfo_lock:
|
|
cache_val, cache_time = self._networkinfo_cache
|
|
if time.time() - cache_time < 60: # seconds
|
|
return cache_val
|
|
val = await self._send_single('getnetworkinfo')
|
|
self._networkinfo_cache = (val, time.time())
|
|
return val
|
|
|
|
async def getmempoolinfo(self):
|
|
"""Return the result of the 'getmempoolinfo' RPC call."""
|
|
async with self._mempoolinfo_lock:
|
|
cache_val, cache_time = self._mempoolinfo_cache
|
|
if time.time() - cache_time < 60: # seconds
|
|
return cache_val
|
|
val = await self._send_single('getmempoolinfo')
|
|
self._mempoolinfo_cache = (val, time.time())
|
|
return val
|
|
|
|
async def getindexinfo(self):
|
|
"""Return the result of the 'getindexinfo' RPC call."""
|
|
# note: not cached as it's not exposed to client sessions
|
|
return await self._send_single('getindexinfo')
|
|
|
|
async def relayfee(self):
|
|
"""Same as getmempoolinfo['minrelaytxfee'].
|
|
The minimum fee required for a transaction to be relayed on by the daemon to the
|
|
network. Doesn't guarantee mempool acceptance."""
|
|
network_info = await self.getnetworkinfo()
|
|
return network_info['relayfee']
|
|
|
|
async def mempool_info(self) -> dict[str, float]:
|
|
mempool_info = await self.getmempoolinfo()
|
|
return {
|
|
# Dynamic minimum fee rate in PLM/kvB for tx to be accepted.
|
|
# Is the maximum of minrelaytxfee and minimum mempool fee
|
|
'mempoolminfee': mempool_info['mempoolminfee'],
|
|
# Minimum relay fee for transactions in PLM/kvB, operator-configurable, static.
|
|
'minrelaytxfee': mempool_info['minrelaytxfee'],
|
|
# Minimum fee rate increment for mempool limiting or replacement in PLM/kvB,
|
|
# operator-configurable, static.
|
|
'incrementalrelayfee': mempool_info['incrementalrelayfee'],
|
|
}
|
|
|
|
async def getrawtransaction(self, txid_hum: str, verbose=False):
|
|
'''Return the serialized raw transaction with the given hash.'''
|
|
# Cast to int because some coin daemons are old and require it
|
|
return await self._send_single('getrawtransaction',
|
|
(txid_hum, int(verbose)))
|
|
|
|
async def getrawtransactions(self, txids_hum: Iterable[str], replace_errs=True) -> Sequence[bytes | None]:
|
|
'''Return the serialized raw transactions with the given hashes.
|
|
|
|
Replaces errors with None by default.'''
|
|
params_iterable = ((txid_hum, 0) for txid_hum in txids_hum)
|
|
txs = await self._send_vector('getrawtransaction', params_iterable,
|
|
replace_errs=replace_errs)
|
|
# Convert hex strings to bytes
|
|
return [hex_to_bytes(tx) if tx else None for tx in txs]
|
|
|
|
async def gettxspendingprevout(self, prev_txid_hum: str, txout_idx: int) -> dict[str, Any]:
|
|
"""Query the daemon to find (if any) the spender of given outpoint."""
|
|
outpoints = [{"txid": prev_txid_hum, "vout": txout_idx}, ]
|
|
options = {"mempool_only": False}
|
|
tx_items = await self._send_single('gettxspendingprevout', (outpoints, options))
|
|
return tx_items[0]
|
|
|
|
async def broadcast_transaction(self, raw_tx: str) -> str:
|
|
'''Broadcast a transaction to the network.'''
|
|
return await self._send_single('sendrawtransaction', (raw_tx, ))
|
|
|
|
async def broadcast_package(self, raw_txs: Sequence[str]):
|
|
"""Broadcast a package of transactions to the network using 'submitpackage'."""
|
|
return await self._send_single('submitpackage', (raw_txs, ))
|
|
|
|
async def testmempoolaccept(self, raw_txs: Sequence[str]):
|
|
"""Query the daemon to test mempool acceptance of txs,
|
|
without adding them to the mempool or broadcasting them.
|
|
"""
|
|
return await self._send_single('testmempoolaccept', (raw_txs, ))
|
|
|
|
async def height(self) -> int:
|
|
'''Query the daemon for its current height.'''
|
|
old_height = self._height
|
|
self._height = await self._send_single('getblockcount')
|
|
if old_height != self._height:
|
|
self.logger.debug(f"new height: {self._height}")
|
|
self.height_changed.set()
|
|
self.height_changed.clear()
|
|
return self._height
|
|
|
|
def cached_height(self) -> Optional[int]:
|
|
'''Return the cached daemon height.
|
|
|
|
If the daemon has not been queried yet this returns None.'''
|
|
return self._height
|