aiorpcx address synchronizer
This commit is contained in:
Janus
@@ -22,102 +22,86 @@
|
||||
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
# SOFTWARE.
|
||||
import traceback
|
||||
import ssl
|
||||
import asyncio
|
||||
from aiorpcx import ClientSession, Request, Notification, TaskGroup
|
||||
from threading import Lock
|
||||
import hashlib
|
||||
import concurrent.futures
|
||||
|
||||
# from .bitcoin import Hash, hash_encode
|
||||
from .transaction import Transaction
|
||||
from .util import ThreadJob, bh2u
|
||||
from .util import ThreadJob, bh2u, PrintError, aiosafe
|
||||
from .bitcoin import address_to_scripthash
|
||||
from .version import ELECTRUM_VERSION, PROTOCOL_VERSION
|
||||
|
||||
|
||||
class Synchronizer(ThreadJob):
|
||||
def history_status(h):
|
||||
if not h:
|
||||
return None
|
||||
status = ''
|
||||
for tx_hash, height in h:
|
||||
status += tx_hash + ':%d:' % height
|
||||
return bh2u(hashlib.sha256(status.encode('ascii')).digest())
|
||||
|
||||
|
||||
|
||||
class NotificationSession(ClientSession):
|
||||
|
||||
def __init__(self, queue, *args, **kwargs):
|
||||
super(NotificationSession, self).__init__(*args, **kwargs)
|
||||
self.queue = queue
|
||||
|
||||
@aiosafe
|
||||
async def handle_request(self, request):
|
||||
if isinstance(request, Notification):
|
||||
if request.method == 'blockchain.scripthash.subscribe':
|
||||
args = request.args
|
||||
await self.queue.put((args[0], args[1]))
|
||||
|
||||
|
||||
class Synchronizer(PrintError):
|
||||
'''The synchronizer keeps the wallet up-to-date with its set of
|
||||
addresses and their transactions. It subscribes over the network
|
||||
to wallet addresses, gets the wallet to generate new addresses
|
||||
when necessary, requests the transaction history of any addresses
|
||||
we don't have the full history of, and requests binary transaction
|
||||
data of any transactions the wallet doesn't have.
|
||||
|
||||
External interface: __init__() and add() member functions.
|
||||
'''
|
||||
|
||||
def __init__(self, wallet, network):
|
||||
def __init__(self, wallet):
|
||||
self.wallet = wallet
|
||||
self.network = network
|
||||
self.new_addresses = set()
|
||||
# Entries are (tx_hash, tx_height) tuples
|
||||
self.requested_tx = {}
|
||||
self.requested_histories = {}
|
||||
self.requested_addrs = set()
|
||||
self.lock = Lock()
|
||||
self.scripthash_to_address = {}
|
||||
# Queues
|
||||
self.add_queue = asyncio.Queue()
|
||||
self.status_queue = asyncio.Queue()
|
||||
|
||||
self.initialized = False
|
||||
self.initialize()
|
||||
|
||||
def parse_response(self, response):
|
||||
if response.get('error'):
|
||||
self.print_error("response error:", response)
|
||||
return None, None
|
||||
return response['params'], response['result']
|
||||
async def send_version(self):
|
||||
r = await self.session.send_request('server.version', [ELECTRUM_VERSION, PROTOCOL_VERSION])
|
||||
|
||||
def is_up_to_date(self):
|
||||
return (not self.requested_tx and not self.requested_histories
|
||||
and not self.requested_addrs)
|
||||
return (not self.requested_addrs and not self.requested_histories)
|
||||
|
||||
def release(self):
|
||||
self.network.unsubscribe(self.on_address_status)
|
||||
def add(self, addr):
|
||||
self.requested_addrs.add(addr)
|
||||
self.add_queue.put_nowait(addr)
|
||||
|
||||
def add(self, address):
|
||||
'''This can be called from the proxy or GUI threads.'''
|
||||
with self.lock:
|
||||
self.new_addresses.add(address)
|
||||
|
||||
def subscribe_to_addresses(self, addresses):
|
||||
if addresses:
|
||||
self.requested_addrs |= addresses
|
||||
self.network.subscribe_to_addresses(addresses, self.on_address_status)
|
||||
|
||||
def get_status(self, h):
|
||||
if not h:
|
||||
return None
|
||||
status = ''
|
||||
for tx_hash, height in h:
|
||||
status += tx_hash + ':%d:' % height
|
||||
return bh2u(hashlib.sha256(status.encode('ascii')).digest())
|
||||
|
||||
def on_address_status(self, response):
|
||||
if self.wallet.synchronizer is None and self.initialized:
|
||||
return # we have been killed, this was just an orphan callback
|
||||
params, result = self.parse_response(response)
|
||||
if not params:
|
||||
return
|
||||
addr = params[0]
|
||||
async def on_address_status(self, addr, status):
|
||||
history = self.wallet.history.get(addr, [])
|
||||
if self.get_status(history) != result:
|
||||
# note that at this point 'result' can be None;
|
||||
# if we had a history for addr but now the server is telling us
|
||||
# there is no history
|
||||
if addr not in self.requested_histories:
|
||||
self.requested_histories[addr] = result
|
||||
self.network.request_address_history(addr, self.on_address_history)
|
||||
# remove addr from list only after it is added to requested_histories
|
||||
if addr in self.requested_addrs: # Notifications won't be in
|
||||
self.requested_addrs.remove(addr)
|
||||
|
||||
def on_address_history(self, response):
|
||||
if self.wallet.synchronizer is None and self.initialized:
|
||||
return # we have been killed, this was just an orphan callback
|
||||
params, result = self.parse_response(response)
|
||||
if not params:
|
||||
if history_status(history) == status:
|
||||
return
|
||||
addr = params[0]
|
||||
try:
|
||||
server_status = self.requested_histories[addr]
|
||||
except KeyError:
|
||||
# note: server_status can be None even if we asked for the history,
|
||||
# so it is not sufficient to test that
|
||||
self.print_error("receiving history (unsolicited)", addr, len(result))
|
||||
# note that at this point 'result' can be None;
|
||||
# if we had a history for addr but now the server is telling us
|
||||
# there is no history
|
||||
if addr in self.requested_histories:
|
||||
return
|
||||
# request address history
|
||||
self.requested_histories[addr] = status
|
||||
h = address_to_scripthash(addr)
|
||||
result = await self.session.send_request("blockchain.scripthash.get_history", [h])
|
||||
self.print_error("receiving history", addr, len(result))
|
||||
hashes = set(map(lambda item: item['tx_hash'], result))
|
||||
hist = list(map(lambda item: (item['tx_hash'], item['height']), result))
|
||||
@@ -128,23 +112,44 @@ class Synchronizer(ThreadJob):
|
||||
if len(hashes) != len(result):
|
||||
self.print_error("error: server history has non-unique txids: %s"% addr)
|
||||
# Check that the status corresponds to what was announced
|
||||
elif self.get_status(hist) != server_status:
|
||||
elif history_status(hist) != status:
|
||||
self.print_error("error: status mismatch: %s" % addr)
|
||||
else:
|
||||
# Store received history
|
||||
self.wallet.receive_history_callback(addr, hist, tx_fees)
|
||||
# Request transactions we don't have
|
||||
self.request_missing_txs(hist)
|
||||
# "hist" is a list of [tx_hash, tx_height] lists
|
||||
transaction_hashes = []
|
||||
for tx_hash, tx_height in hist:
|
||||
if tx_hash in self.requested_tx:
|
||||
continue
|
||||
if tx_hash in self.wallet.transactions:
|
||||
continue
|
||||
transaction_hashes.append(tx_hash)
|
||||
self.requested_tx[tx_hash] = tx_height
|
||||
|
||||
for tx_hash in transaction_hashes:
|
||||
await self.get_transaction(tx_hash)
|
||||
|
||||
# Remove request; this allows up_to_date to be True
|
||||
self.requested_histories.pop(addr)
|
||||
|
||||
def on_tx_response(self, response):
|
||||
if self.wallet.synchronizer is None and self.initialized:
|
||||
return # we have been killed, this was just an orphan callback
|
||||
params, result = self.parse_response(response)
|
||||
if not params:
|
||||
return
|
||||
tx_hash = params[0]
|
||||
async def request_missing_txs(self, hist):
|
||||
# "hist" is a list of [tx_hash, tx_height] lists
|
||||
transaction_hashes = []
|
||||
for tx_hash, tx_height in hist:
|
||||
if tx_hash in self.requested_tx:
|
||||
continue
|
||||
if tx_hash in self.wallet.transactions:
|
||||
continue
|
||||
transaction_hashes.append(tx_hash)
|
||||
self.requested_tx[tx_hash] = tx_height
|
||||
|
||||
for tx_hash in transaction_hashes:
|
||||
await self.get_transaction(tx_hash)
|
||||
|
||||
async def get_transaction(self, tx_hash):
|
||||
result = await self.session.send_request('blockchain.transaction.get', [tx_hash])
|
||||
tx = Transaction(result)
|
||||
try:
|
||||
tx.deserialize()
|
||||
@@ -160,54 +165,44 @@ class Synchronizer(ThreadJob):
|
||||
self.print_error("received tx %s height: %d bytes: %d" %
|
||||
(tx_hash, tx_height, len(tx.raw)))
|
||||
# callbacks
|
||||
self.network.trigger_callback('new_transaction', tx)
|
||||
if not self.requested_tx:
|
||||
self.network.trigger_callback('updated')
|
||||
self.wallet.network.trigger_callback('new_transaction', tx)
|
||||
|
||||
def request_missing_txs(self, hist):
|
||||
# "hist" is a list of [tx_hash, tx_height] lists
|
||||
transaction_hashes = []
|
||||
for tx_hash, tx_height in hist:
|
||||
if tx_hash in self.requested_tx:
|
||||
continue
|
||||
if tx_hash in self.wallet.transactions:
|
||||
continue
|
||||
transaction_hashes.append(tx_hash)
|
||||
self.requested_tx[tx_hash] = tx_height
|
||||
async def subscribe_to_address(self, addr):
|
||||
h = address_to_scripthash(addr)
|
||||
self.scripthash_to_address[h] = addr
|
||||
status = await self.session.send_request('blockchain.scripthash.subscribe', [h])
|
||||
await self.status_queue.put((h, status))
|
||||
self.requested_addrs.remove(addr)
|
||||
|
||||
self.network.get_transactions(transaction_hashes, self.on_tx_response)
|
||||
@aiosafe
|
||||
async def send_subscriptions(self):
|
||||
async with TaskGroup() as group:
|
||||
while True:
|
||||
addr = await self.add_queue.get()
|
||||
await group.spawn(self.subscribe_to_address(addr))
|
||||
|
||||
def initialize(self):
|
||||
'''Check the initial state of the wallet. Subscribe to all its
|
||||
addresses, and request any transactions in its address history
|
||||
we don't have.
|
||||
'''
|
||||
for history in self.wallet.history.values():
|
||||
# Old electrum servers returned ['*'] when all history for
|
||||
# the address was pruned. This no longer happens but may
|
||||
# remain in old wallets.
|
||||
if history == ['*']:
|
||||
continue
|
||||
self.request_missing_txs(history)
|
||||
@aiosafe
|
||||
async def handle_status(self):
|
||||
async with TaskGroup() as group:
|
||||
while True:
|
||||
h, status = await self.status_queue.get()
|
||||
addr = self.scripthash_to_address[h]
|
||||
await group.spawn(self.on_address_status(addr, status))
|
||||
|
||||
if self.requested_tx:
|
||||
self.print_error("missing tx", self.requested_tx)
|
||||
self.subscribe_to_addresses(set(self.wallet.get_addresses()))
|
||||
self.initialized = True
|
||||
|
||||
def run(self):
|
||||
'''Called from the network proxy thread main loop.'''
|
||||
# 1. Create new addresses
|
||||
self.wallet.synchronize()
|
||||
|
||||
# 2. Subscribe to new addresses
|
||||
with self.lock:
|
||||
addresses = self.new_addresses
|
||||
self.new_addresses = set()
|
||||
self.subscribe_to_addresses(addresses)
|
||||
|
||||
# 3. Detect if situation has changed
|
||||
up_to_date = self.is_up_to_date()
|
||||
if up_to_date != self.wallet.is_up_to_date():
|
||||
self.wallet.set_up_to_date(up_to_date)
|
||||
self.network.trigger_callback('updated')
|
||||
@aiosafe
|
||||
async def main(self):
|
||||
conn = self.wallet.network.default_server
|
||||
host, port, protocol = conn.split(':')
|
||||
sslc = ssl.SSLContext(ssl.PROTOCOL_TLS) if protocol == 's' else None
|
||||
async with NotificationSession(self.status_queue, host, int(port), ssl=sslc) as session:
|
||||
self.session = session
|
||||
await self.send_version()
|
||||
self.wallet.synchronizer = self
|
||||
for addr in self.wallet.get_addresses(): self.add(addr)
|
||||
while True:
|
||||
await asyncio.sleep(1)
|
||||
self.wallet.synchronize()
|
||||
up_to_date = self.is_up_to_date()
|
||||
if up_to_date != self.wallet.is_up_to_date():
|
||||
self.wallet.set_up_to_date(up_to_date)
|
||||
self.wallet.network.trigger_callback('updated')
|
||||
|
||||
Reference in New Issue
Block a user