Files
purple-electrumwallet/electrum/address_synchronizer.py
T

954 lines
42 KiB
Python
Raw Normal View History

2018-07-18 11:18:57 +02:00
# Electrum - lightweight Bitcoin client
# Copyright (C) 2018 The Electrum Developers
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation files
# (the "Software"), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# 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.
2018-04-12 14:14:41 +02:00
import asyncio
2018-07-18 11:18:57 +02:00
import threading
import itertools
from collections import defaultdict
from typing import TYPE_CHECKING, Dict, Optional, Set, Tuple, NamedTuple, Sequence, List
2018-07-18 11:18:57 +02:00
2022-04-12 09:53:30 +02:00
from .crypto import sha256
2020-04-14 16:12:47 +02:00
from . import bitcoin, util
2019-10-23 17:09:41 +02:00
from .bitcoin import COINBASE_MATURITY
2022-02-08 12:34:49 +01:00
from .util import profiler, bfh, TxMinedInfo, UnrelatedTransactionException, with_lock, OldTaskGroup
2019-11-07 02:24:16 +01:00
from .transaction import Transaction, TxOutput, TxInput, PartialTxInput, TxOutpoint, PartialTransaction
2018-07-18 11:18:57 +02:00
from .synchronizer import Synchronizer
from .verifier import SPV
from .blockchain import hash_header, Blockchain
from .i18n import _
2019-04-26 18:52:26 +02:00
from .logging import Logger
2022-06-16 12:05:05 +02:00
from .util import EventListener, event_listener
2018-10-25 23:01:53 +02:00
if TYPE_CHECKING:
from .network import Network
from .wallet_db import WalletDB
from .simple_config import SimpleConfig
2018-10-22 16:41:25 +02:00
2018-07-18 11:18:57 +02:00
TX_HEIGHT_FUTURE = -3
2018-07-18 11:18:57 +02:00
TX_HEIGHT_LOCAL = -2
TX_HEIGHT_UNCONF_PARENT = -1
TX_HEIGHT_UNCONFIRMED = 0
class HistoryItem(NamedTuple):
txid: str
tx_mined_status: TxMinedInfo
2020-10-18 22:14:52 +02:00
delta: int
fee: Optional[int]
2020-10-18 22:14:52 +02:00
balance: int
2022-06-16 12:05:05 +02:00
class AddressSynchronizer(Logger, EventListener):
""" address database """
2018-07-18 11:18:57 +02:00
network: Optional['Network']
asyncio_loop: Optional['asyncio.AbstractEventLoop'] = None
synchronizer: Optional['Synchronizer']
verifier: Optional['SPV']
def __init__(self, db: 'WalletDB', config: 'SimpleConfig', *, name: str = None):
2019-07-03 08:46:00 +02:00
self.db = db
2022-06-07 16:52:09 +02:00
self.config = config
self.name = name
self.network = None
2019-04-26 18:52:26 +02:00
Logger.__init__(self)
2018-10-03 17:13:46 +02:00
# verifier (SPV) and synchronizer are started in start_network
self.synchronizer = None
self.verifier = None
2018-07-18 11:18:57 +02:00
# locks: if you need to take multiple ones, acquire them in the order they are defined here!
self.lock = threading.RLock()
self.transaction_lock = threading.RLock()
self.future_tx = {} # type: Dict[str, int] # txid -> wanted height
# Txs the server claims are mined but still pending verification:
self.unverified_tx = defaultdict(int) # type: Dict[str, int] # txid -> height. Access with self.lock.
# Txs the server claims are in the mempool:
self.unconfirmed_tx = defaultdict(int) # type: Dict[str, int] # txid -> height. Access with self.lock.
# thread local storage for caching stuff
self.threadlocal_cache = threading.local()
2018-07-19 18:16:23 +02:00
2022-04-12 09:53:30 +02:00
self._get_balance_cache = {}
2019-03-04 22:19:33 +01:00
2018-07-19 18:16:23 +02:00
self.load_and_cleanup()
def diagnostic_name(self):
return self.name or ""
def with_transaction_lock(func):
2020-10-18 22:21:06 +02:00
def func_wrapper(self: 'AddressSynchronizer', *args, **kwargs):
with self.transaction_lock:
return func(self, *args, **kwargs)
return func_wrapper
2018-07-19 18:16:23 +02:00
def load_and_cleanup(self):
2018-07-18 11:18:57 +02:00
self.load_local_history()
self.check_history()
2018-07-18 11:18:57 +02:00
self.load_unverified_transactions()
self.remove_local_transactions_we_dont_have()
def is_mine(self, address: Optional[str]) -> bool:
"""Returns whether an address is in our set
Note: This class has a larget set of addresses than the wallet
"""
if not address: return False
return self.db.is_addr_in_history(address)
def get_addresses(self):
return sorted(self.db.get_history())
def get_address_history(self, addr: str) -> Sequence[Tuple[str, int]]:
"""Returns the history for the address, in the format that would be returned by a server.
Note: The difference between db.get_addr_history and this method is that
db.get_addr_history stores the response from a server, so it only includes txns
a server sees, i.e. that does not contain local and future txns.
"""
h = []
# we need self.transaction_lock but get_tx_height will take self.lock
# so we need to take that too here, to enforce order of locks
with self.lock, self.transaction_lock:
related_txns = self._history_local.get(addr, set())
for tx_hash in related_txns:
tx_height = self.get_tx_height(tx_hash).height
h.append((tx_hash, tx_height))
return h
def get_address_history_len(self, addr: str) -> int:
2018-08-14 21:54:11 +02:00
"""Return number of transactions where address is involved."""
return len(self._history_local.get(addr, ()))
2019-10-23 17:09:41 +02:00
def get_txin_address(self, txin: TxInput) -> Optional[str]:
if isinstance(txin, PartialTxInput):
if txin.address:
return txin.address
prevout_hash = txin.prevout.txid.hex()
prevout_n = txin.prevout.out_idx
2019-09-11 21:47:44 +02:00
for addr in self.db.get_txo_addresses(prevout_hash):
d = self.db.get_txo_addr(prevout_hash, addr)
if prevout_n in d:
return addr
tx = self.db.get_transaction(prevout_hash)
if tx:
return tx.outputs()[prevout_n].address
return None
def get_txin_value(self, txin: TxInput, *, address: str = None) -> Optional[int]:
if txin.value_sats() is not None:
return txin.value_sats()
prevout_hash = txin.prevout.txid.hex()
prevout_n = txin.prevout.out_idx
2020-10-16 21:25:11 +02:00
if address is None:
address = self.get_txin_address(txin)
if address:
d = self.db.get_txo_addr(prevout_hash, address)
try:
v, cb = d[prevout_n]
return v
except KeyError:
pass
tx = self.db.get_transaction(prevout_hash)
if tx:
return tx.outputs()[prevout_n].value
return None
2018-07-18 11:18:57 +02:00
def load_unverified_transactions(self):
# review transactions that are in the history
for addr in self.db.get_history():
hist = self.db.get_addr_history(addr)
2018-07-18 11:18:57 +02:00
for tx_hash, tx_height in hist:
# add it in case it was previously unconfirmed
self.add_unverified_or_unconfirmed_tx(tx_hash, tx_height)
2018-07-18 11:18:57 +02:00
def start_network(self, network: Optional['Network']) -> None:
2018-07-18 11:18:57 +02:00
self.network = network
if self.network is not None:
2018-10-03 17:13:46 +02:00
self.synchronizer = Synchronizer(self)
self.verifier = SPV(self.network, self)
self.asyncio_loop = network.asyncio_loop
2022-06-16 12:05:05 +02:00
self.register_callbacks()
2019-03-04 22:19:33 +01:00
2022-06-16 12:05:05 +02:00
@event_listener
def on_event_blockchain_updated(self, *args):
2022-04-12 09:53:30 +02:00
self._get_balance_cache = {} # invalidate cache
2018-07-18 11:18:57 +02:00
async def stop(self):
2018-07-18 11:18:57 +02:00
if self.network:
try:
2022-02-08 12:34:49 +01:00
async with OldTaskGroup() as group:
if self.synchronizer:
await group.spawn(self.synchronizer.stop())
if self.verifier:
await group.spawn(self.verifier.stop())
finally: # even if we get cancelled
2018-10-03 17:13:46 +02:00
self.synchronizer = None
self.verifier = None
2022-06-16 12:05:05 +02:00
self.unregister_callbacks()
self.db.put('stored_height', self.get_local_height())
2018-07-18 11:18:57 +02:00
def add_address(self, address):
if address not in self.db.history:
self.db.history[address] = []
2018-07-18 11:18:57 +02:00
if self.synchronizer:
self.synchronizer.add(address)
2022-12-21 15:23:11 +00:00
self.up_to_date_changed()
2018-07-18 11:18:57 +02:00
2019-10-23 17:09:41 +02:00
def get_conflicting_transactions(self, tx_hash, tx: Transaction, include_self=False):
2018-07-18 11:18:57 +02:00
"""Returns a set of transaction hashes from the wallet history that are
directly conflicting with tx, i.e. they have common outpoints being
2018-09-12 16:17:10 +02:00
spent with tx.
include_self specifies whether the tx itself should be reported as a
conflict (if already in wallet history)
2018-07-18 11:18:57 +02:00
"""
conflicting_txns = set()
with self.transaction_lock:
for txin in tx.inputs():
2020-01-02 00:43:49 +01:00
if txin.is_coinbase_input():
2018-07-18 11:18:57 +02:00
continue
2019-10-23 17:09:41 +02:00
prevout_hash = txin.prevout.txid.hex()
prevout_n = txin.prevout.out_idx
spending_tx_hash = self.db.get_spent_outpoint(prevout_hash, prevout_n)
2018-07-18 11:18:57 +02:00
if spending_tx_hash is None:
continue
# this outpoint has already been spent, by spending_tx
2019-06-15 03:51:11 +02:00
# annoying assert that has revealed several bugs over time:
assert self.db.get_transaction(spending_tx_hash), "spending tx not in wallet db"
2018-07-18 11:18:57 +02:00
conflicting_txns |= {spending_tx_hash}
if tx_hash in conflicting_txns:
2018-07-18 11:18:57 +02:00
# this tx is already in history, so it conflicts with itself
if len(conflicting_txns) > 1:
raise Exception('Found conflicting transactions already in wallet history.')
2018-09-12 16:17:10 +02:00
if not include_self:
conflicting_txns -= {tx_hash}
2018-07-18 11:18:57 +02:00
return conflicting_txns
def get_transaction(self, txid: str) -> Optional[Transaction]:
tx = self.db.get_transaction(txid)
2023-01-26 11:13:35 +01:00
if tx:
# add verified tx info
tx.deserialize()
for txin in tx._inputs:
tx_height, tx_pos = self.get_txpos(txin.prevout.txid.hex())
txin.block_height = tx_height
txin.block_txpos = tx_pos
return tx
def add_transaction(self, tx: Transaction, *, allow_unrelated=False, is_new=True) -> bool:
"""
Returns whether the tx was successfully added to the wallet history.
Note that a transaction may need to be added several times, if our
list of addresses has increased. This will return True even if the
transaction was already in self.db.
"""
2018-07-18 11:18:57 +02:00
assert tx, tx
# note: tx.is_complete() is not necessarily True; tx might be partial
# but it *needs* to have a txid:
2019-11-23 12:46:43 +01:00
tx_hash = tx.txid()
if tx_hash is None:
raise Exception("cannot add tx without txid to wallet history")
2018-07-18 11:18:57 +02:00
# we need self.transaction_lock but get_tx_height will take self.lock
# so we need to take that too here, to enforce order of locks
with self.lock, self.transaction_lock:
# NOTE: returning if tx in self.transactions might seem like a good idea
# BUT we track is_mine inputs in a txn, and during subsequent calls
# of add_transaction tx, we might learn of more-and-more inputs of
# being is_mine, as we roll the gap_limit forward
2020-01-02 00:43:49 +01:00
is_coinbase = tx.inputs()[0].is_coinbase_input()
tx_height = self.get_tx_height(tx_hash).height
2018-07-18 11:18:57 +02:00
if not allow_unrelated:
# note that during sync, if the transactions are not properly sorted,
# it could happen that we think tx is unrelated but actually one of the inputs is is_mine.
# this is the main motivation for allow_unrelated
is_mine = any([self.is_mine(self.get_txin_address(txin)) for txin in tx.inputs()])
2021-06-08 16:45:30 +02:00
is_for_me = any([self.is_mine(txo.address) for txo in tx.outputs()])
2018-07-18 11:18:57 +02:00
if not is_mine and not is_for_me:
raise UnrelatedTransactionException()
# Find all conflicting transactions.
# In case of a conflict,
# 1. confirmed > mempool > local
# 2. this new txn has priority over existing ones
# When this method exits, there must NOT be any conflict, so
# either keep this txn and remove all conflicting (along with dependencies)
# or drop this txn
conflicting_txns = self.get_conflicting_transactions(tx_hash, tx)
2018-07-18 11:18:57 +02:00
if conflicting_txns:
existing_mempool_txn = any(
self.get_tx_height(tx_hash2).height in (TX_HEIGHT_UNCONFIRMED, TX_HEIGHT_UNCONF_PARENT)
2018-07-18 11:18:57 +02:00
for tx_hash2 in conflicting_txns)
existing_confirmed_txn = any(
self.get_tx_height(tx_hash2).height > 0
2018-07-18 11:18:57 +02:00
for tx_hash2 in conflicting_txns)
if existing_confirmed_txn and tx_height <= 0:
# this is a non-confirmed tx that conflicts with confirmed txns; drop.
return False
if existing_mempool_txn and tx_height == TX_HEIGHT_LOCAL:
# this is a local tx that conflicts with non-local txns; drop.
return False
# keep this txn and remove all conflicting
for tx_hash2 in conflicting_txns:
2018-07-18 11:18:57 +02:00
self.remove_transaction(tx_hash2)
# add inputs
def add_value_from_prev_output():
# note: this takes linear time in num is_mine outputs of prev_tx
addr = self.get_txin_address(txi)
if addr and self.is_mine(addr):
outputs = self.db.get_txo_addr(prevout_hash, addr)
try:
v, is_cb = outputs[prevout_n]
except KeyError:
pass
else:
self.db.add_txi_addr(tx_hash, addr, ser, v)
2022-04-12 09:53:30 +02:00
self._get_balance_cache.clear() # invalidate cache
2018-07-18 11:18:57 +02:00
for txi in tx.inputs():
2020-01-02 00:43:49 +01:00
if txi.is_coinbase_input():
2018-07-18 11:18:57 +02:00
continue
2019-10-23 17:09:41 +02:00
prevout_hash = txi.prevout.txid.hex()
prevout_n = txi.prevout.out_idx
ser = txi.prevout.to_str()
self.db.set_spent_outpoint(prevout_hash, prevout_n, tx_hash)
2018-07-18 11:18:57 +02:00
add_value_from_prev_output()
# add outputs
for n, txo in enumerate(tx.outputs()):
2019-03-03 17:34:03 +01:00
v = txo.value
2018-07-18 11:18:57 +02:00
ser = tx_hash + ':%d'%n
scripthash = bitcoin.script_to_scripthash(txo.scriptpubkey.hex())
self.db.add_prevout_by_scripthash(scripthash, prevout=TxOutpoint.from_str(ser), value=v)
2021-06-08 16:45:30 +02:00
addr = txo.address
2018-07-18 11:18:57 +02:00
if addr and self.is_mine(addr):
self.db.add_txo_addr(tx_hash, addr, n, v, is_coinbase)
2022-04-12 09:53:30 +02:00
self._get_balance_cache.clear() # invalidate cache
2018-07-18 11:18:57 +02:00
# give v to txi that spends me
next_tx = self.db.get_spent_outpoint(tx_hash, n)
2018-07-18 11:18:57 +02:00
if next_tx is not None:
self.db.add_txi_addr(next_tx, addr, ser, v)
2018-07-18 11:18:57 +02:00
self._add_tx_to_local_history(next_tx)
# add to local history
self._add_tx_to_local_history(tx_hash)
# save
self.db.add_transaction(tx_hash, tx)
self.db.add_num_inputs_to_tx(tx_hash, len(tx.inputs()))
if is_new:
util.trigger_callback('adb_added_tx', self, tx_hash)
2018-07-18 11:18:57 +02:00
return True
def remove_transaction(self, tx_hash: str) -> None:
"""Removes a transaction AND all its dependents/children
from the wallet history.
"""
with self.lock, self.transaction_lock:
to_remove = {tx_hash}
to_remove |= self.get_depending_transactions(tx_hash)
for txid in to_remove:
self._remove_transaction(txid)
def _remove_transaction(self, tx_hash: str) -> None:
"""Removes a single transaction from the wallet history, and attempts
to undo all effects of the tx (spending inputs, creating outputs, etc).
"""
2018-07-18 11:18:57 +02:00
def remove_from_spent_outpoints():
# undo spends in spent_outpoints
if tx is not None:
# if we have the tx, this branch is faster
2018-07-18 11:18:57 +02:00
for txin in tx.inputs():
2020-01-02 00:43:49 +01:00
if txin.is_coinbase_input():
2018-07-18 11:18:57 +02:00
continue
2019-10-23 17:09:41 +02:00
prevout_hash = txin.prevout.txid.hex()
prevout_n = txin.prevout.out_idx
self.db.remove_spent_outpoint(prevout_hash, prevout_n)
else:
# expensive but always works
2019-02-28 15:47:42 +01:00
for prevout_hash, prevout_n in self.db.list_spent_outpoints():
spending_txid = self.db.get_spent_outpoint(prevout_hash, prevout_n)
if spending_txid == tx_hash:
self.db.remove_spent_outpoint(prevout_hash, prevout_n)
2018-07-18 11:18:57 +02:00
with self.lock, self.transaction_lock:
2019-05-06 23:03:19 +02:00
self.logger.info(f"removing tx from history {tx_hash}")
tx = self.db.remove_transaction(tx_hash)
2018-07-18 11:18:57 +02:00
remove_from_spent_outpoints()
self._remove_tx_from_local_history(tx_hash)
2019-09-11 21:47:44 +02:00
for addr in itertools.chain(self.db.get_txi_addresses(tx_hash), self.db.get_txo_addresses(tx_hash)):
2022-04-12 09:53:30 +02:00
self._get_balance_cache.clear() # invalidate cache
self.db.remove_txi(tx_hash)
self.db.remove_txo(tx_hash)
self.db.remove_tx_fee(tx_hash)
self.db.remove_verified_tx(tx_hash)
self.unverified_tx.pop(tx_hash, None)
self.unconfirmed_tx.pop(tx_hash, None)
if tx:
for idx, txo in enumerate(tx.outputs()):
scripthash = bitcoin.script_to_scripthash(txo.scriptpubkey.hex())
prevout = TxOutpoint(bfh(tx_hash), idx)
self.db.remove_prevout_by_scripthash(scripthash, prevout=prevout, value=txo.value)
2018-07-18 11:18:57 +02:00
def get_depending_transactions(self, tx_hash: str) -> Set[str]:
"""Returns all (grand-)children of tx_hash in this wallet."""
2019-02-20 18:01:43 +01:00
with self.transaction_lock:
children = set()
for n in self.db.get_spent_outpoints(tx_hash):
other_hash = self.db.get_spent_outpoint(tx_hash, n)
2019-02-20 18:01:43 +01:00
children.add(other_hash)
children |= self.get_depending_transactions(other_hash)
return children
def receive_tx_callback(self, tx_hash: str, tx: Transaction, tx_height: int) -> None:
self.add_unverified_or_unconfirmed_tx(tx_hash, tx_height)
2019-11-23 12:46:43 +01:00
self.add_transaction(tx, allow_unrelated=True)
2018-07-18 11:18:57 +02:00
def receive_history_callback(self, addr: str, hist, tx_fees: Dict[str, int]):
2018-07-18 11:18:57 +02:00
with self.lock:
old_hist = self.get_address_history(addr)
for tx_hash, height in old_hist:
if (tx_hash, height) not in hist:
# make tx local
self.unverified_tx.pop(tx_hash, None)
self.unconfirmed_tx.pop(tx_hash, None)
self.db.remove_verified_tx(tx_hash)
2018-07-18 11:18:57 +02:00
if self.verifier:
self.verifier.remove_spv_proof_for_tx(tx_hash)
self.db.set_addr_history(addr, hist)
2018-07-18 11:18:57 +02:00
for tx_hash, tx_height in hist:
# add it in case it was previously unconfirmed
self.add_unverified_or_unconfirmed_tx(tx_hash, tx_height)
2018-07-18 11:18:57 +02:00
# if addr is new, we have to recompute txi and txo
tx = self.db.get_transaction(tx_hash)
2018-07-18 11:18:57 +02:00
if tx is None:
continue
self.add_transaction(tx, allow_unrelated=True, is_new=False)
2018-07-18 11:18:57 +02:00
# Store fees
for tx_hash, fee_sat in tx_fees.items():
self.db.add_tx_fee_from_server(tx_hash, fee_sat)
2018-07-18 11:18:57 +02:00
@profiler
def load_local_history(self):
self._history_local = {} # type: Dict[str, Set[str]] # address -> set(txid)
self._address_history_changed_events = defaultdict(asyncio.Event) # address -> Event
2019-02-28 15:47:42 +01:00
for txid in itertools.chain(self.db.list_txi(), self.db.list_txo()):
2018-07-18 11:18:57 +02:00
self._add_tx_to_local_history(txid)
@profiler
def check_history(self):
hist_addrs_mine = list(filter(lambda k: self.is_mine(k), self.db.get_history()))
hist_addrs_not_mine = list(filter(lambda k: not self.is_mine(k), self.db.get_history()))
for addr in hist_addrs_not_mine:
self.db.remove_addr_history(addr)
for addr in hist_addrs_mine:
hist = self.db.get_addr_history(addr)
for tx_hash, tx_height in hist:
2019-09-11 21:47:44 +02:00
if self.db.get_txi_addresses(tx_hash) or self.db.get_txo_addresses(tx_hash):
continue
tx = self.db.get_transaction(tx_hash)
if tx is not None:
2019-11-23 12:46:43 +01:00
self.add_transaction(tx, allow_unrelated=True)
2018-07-18 11:18:57 +02:00
def remove_local_transactions_we_dont_have(self):
2019-02-28 15:47:42 +01:00
for txid in itertools.chain(self.db.list_txi(), self.db.list_txo()):
tx_height = self.get_tx_height(txid).height
if tx_height == TX_HEIGHT_LOCAL and not self.db.get_transaction(txid):
2018-07-18 11:18:57 +02:00
self.remove_transaction(txid)
def clear_history(self):
with self.lock:
with self.transaction_lock:
self.db.clear_history()
2020-10-09 16:22:59 +02:00
self._history_local.clear()
2022-04-12 09:53:30 +02:00
self._get_balance_cache.clear() # invalidate cache
2018-07-18 11:18:57 +02:00
def get_txpos(self, tx_hash: str) -> Tuple[int, int]:
"""Returns (height, txpos) tuple, even if the tx is unverified."""
with self.lock:
verified_tx_mined_info = self.db.get_verified_tx(tx_hash)
if verified_tx_mined_info:
height = verified_tx_mined_info.height
txpos = verified_tx_mined_info.txpos
assert height > 0, height
assert txpos is not None
return height, txpos
elif tx_hash in self.unverified_tx:
height = self.unverified_tx[tx_hash]
assert height > 0, height
return height, -1
elif tx_hash in self.unconfirmed_tx:
height = self.unconfirmed_tx[tx_hash]
assert height <= 0, height
return (10**9 - height), -1
else:
return (10**9 + 1), -1
def with_local_height_cached(func):
# get local height only once, as it's relatively expensive.
# take care that nested calls work as expected
def f(self, *args, **kwargs):
orig_val = getattr(self.threadlocal_cache, 'local_height', None)
self.threadlocal_cache.local_height = orig_val or self.get_local_height()
try:
return func(self, *args, **kwargs)
finally:
self.threadlocal_cache.local_height = orig_val
return f
2020-10-18 22:21:06 +02:00
@with_lock
@with_transaction_lock
@with_local_height_cached
def get_history(self, domain) -> Sequence[HistoryItem]:
2018-07-18 11:18:57 +02:00
domain = set(domain)
# 1. Get the history of each address in the domain, maintain the
# delta of a tx as the sum of its deltas on domain addresses
2020-10-18 22:14:52 +02:00
tx_deltas = defaultdict(int) # type: Dict[str, int]
2018-07-18 11:18:57 +02:00
for addr in domain:
h = self.get_address_history(addr)
for tx_hash, height in h:
2020-10-18 22:14:52 +02:00
tx_deltas[tx_hash] += self.get_tx_delta(tx_hash, addr)
2018-07-18 11:18:57 +02:00
# 2. create sorted history
history = []
for tx_hash in tx_deltas:
delta = tx_deltas[tx_hash]
tx_mined_status = self.get_tx_height(tx_hash)
2019-09-12 12:26:49 +02:00
fee = self.get_tx_fee(tx_hash)
history.append((tx_hash, tx_mined_status, delta, fee))
2022-05-07 18:46:28 +02:00
history.sort(key = lambda x: self.get_txpos(x[0]))
2018-07-18 11:18:57 +02:00
# 3. add balance
h2 = []
2022-05-07 18:46:28 +02:00
balance = 0
for tx_hash, tx_mined_status, delta, fee in history:
2022-05-07 18:46:28 +02:00
balance += delta
h2.append(HistoryItem(
txid=tx_hash,
tx_mined_status=tx_mined_status,
delta=delta,
fee=fee,
balance=balance))
# sanity check
c, u, x = self.get_balance(domain)
2022-05-07 18:46:28 +02:00
if balance != c + u + x:
2022-12-03 11:09:51 +01:00
self.logger.error(f'sanity check failed! c={c},u={u},x={x} while history balance={balance}')
2020-10-18 22:21:06 +02:00
raise Exception("wallet.get_history() failed balance sanity-check")
2018-07-18 11:18:57 +02:00
return h2
def _add_tx_to_local_history(self, txid):
with self.transaction_lock:
2019-09-11 21:47:44 +02:00
for addr in itertools.chain(self.db.get_txi_addresses(txid), self.db.get_txo_addresses(txid)):
2018-07-18 11:18:57 +02:00
cur_hist = self._history_local.get(addr, set())
cur_hist.add(txid)
self._history_local[addr] = cur_hist
self._mark_address_history_changed(addr)
2018-07-18 11:18:57 +02:00
def _remove_tx_from_local_history(self, txid):
with self.transaction_lock:
2019-09-11 21:47:44 +02:00
for addr in itertools.chain(self.db.get_txi_addresses(txid), self.db.get_txo_addresses(txid)):
2018-07-18 11:18:57 +02:00
cur_hist = self._history_local.get(addr, set())
try:
cur_hist.remove(txid)
except KeyError:
pass
else:
self._history_local[addr] = cur_hist
self._mark_address_history_changed(addr)
2018-07-18 11:18:57 +02:00
def _mark_address_history_changed(self, addr: str) -> None:
def set_and_clear():
event = self._address_history_changed_events[addr]
# history for this address changed, wake up coroutines:
event.set()
# clear event immediately so that coroutines can wait() for the next change:
event.clear()
if self.asyncio_loop:
self.asyncio_loop.call_soon_threadsafe(set_and_clear)
async def wait_for_address_history_to_change(self, addr: str) -> None:
"""Wait until the server tells us about a new transaction related to addr.
Unconfirmed and confirmed transactions are not distinguished, and so e.g. SPV
is not taken into account.
"""
assert self.is_mine(addr), "address needs to be is_mine to be watched"
await self._address_history_changed_events[addr].wait()
def add_unverified_or_unconfirmed_tx(self, tx_hash, tx_height):
if self.db.is_in_verified_tx(tx_hash):
if tx_height <= 0:
# tx was previously SPV-verified but now in mempool (probably reorg)
with self.lock:
self.db.remove_verified_tx(tx_hash)
self.unconfirmed_tx[tx_hash] = tx_height
if self.verifier:
self.verifier.remove_spv_proof_for_tx(tx_hash)
else:
2018-07-18 11:18:57 +02:00
with self.lock:
if tx_height > 0:
self.unverified_tx[tx_hash] = tx_height
else:
self.unconfirmed_tx[tx_hash] = tx_height
2018-07-18 11:18:57 +02:00
2018-09-17 03:35:25 +02:00
def remove_unverified_tx(self, tx_hash, tx_height):
with self.lock:
new_height = self.unverified_tx.get(tx_hash)
if new_height == tx_height:
self.unverified_tx.pop(tx_hash, None)
def add_verified_tx(self, tx_hash: str, info: TxMinedInfo):
2018-07-18 11:18:57 +02:00
# Remove from the unverified map and add to the verified map
with self.lock:
self.unverified_tx.pop(tx_hash, None)
self.db.add_verified_tx(tx_hash, info)
util.trigger_callback('adb_added_verified_tx', self, tx_hash)
2018-07-18 11:18:57 +02:00
def get_unverified_txs(self) -> Dict[str, int]:
2018-07-18 11:18:57 +02:00
'''Returns a map from tx hash to transaction height'''
with self.lock:
return dict(self.unverified_tx) # copy
def undo_verifications(self, blockchain: Blockchain, above_height: int) -> Set[str]:
2018-07-18 11:18:57 +02:00
'''Used by the verifier when a reorg has happened'''
txs = set()
with self.lock:
2019-02-28 15:47:42 +01:00
for tx_hash in self.db.list_verified_tx():
info = self.db.get_verified_tx(tx_hash)
tx_height = info.height
if tx_height > above_height:
2018-07-18 11:18:57 +02:00
header = blockchain.read_header(tx_height)
if not header or hash_header(header) != info.header_hash:
self.db.remove_verified_tx(tx_hash)
# NOTE: we should add these txns to self.unverified_tx,
# but with what height?
# If on the new fork after the reorg, the txn is at the
# same height, we will not get a status update for the
# address. If the txn is not mined or at a diff height,
# we should get a status update. Unless we put tx into
# unverified_tx, it will turn into local. So we put it
# into unverified_tx with the old height, and if we get
# a status update, that will overwrite it.
self.unverified_tx[tx_hash] = tx_height
2018-07-18 11:18:57 +02:00
txs.add(tx_hash)
for tx_hash in txs:
util.trigger_callback('adb_removed_verified_tx', self, tx_hash)
2018-07-18 11:18:57 +02:00
return txs
2019-09-08 18:26:04 +02:00
def get_local_height(self) -> int:
2018-07-18 11:18:57 +02:00
""" return last known height if we are offline """
cached_local_height = getattr(self.threadlocal_cache, 'local_height', None)
if cached_local_height is not None:
return cached_local_height
2019-07-03 08:46:00 +02:00
return self.network.get_local_height() if self.network else self.db.get('stored_height', 0)
2018-07-18 11:18:57 +02:00
def set_future_tx(self, txid:str, wanted_height: int):
with self.lock:
self.future_tx[txid] = wanted_height
def get_tx_height(self, tx_hash: str) -> TxMinedInfo:
2020-02-24 18:52:33 +01:00
if tx_hash is None: # ugly backwards compat...
return TxMinedInfo(height=TX_HEIGHT_LOCAL, conf=0)
2018-07-18 11:18:57 +02:00
with self.lock:
verified_tx_mined_info = self.db.get_verified_tx(tx_hash)
if verified_tx_mined_info:
conf = max(self.get_local_height() - verified_tx_mined_info.height + 1, 0)
return verified_tx_mined_info._replace(conf=conf)
2018-07-18 11:18:57 +02:00
elif tx_hash in self.unverified_tx:
height = self.unverified_tx[tx_hash]
return TxMinedInfo(height=height, conf=0)
elif tx_hash in self.unconfirmed_tx:
height = self.unconfirmed_tx[tx_hash]
return TxMinedInfo(height=height, conf=0)
elif tx_hash in self.future_tx:
num_blocks_remainining = self.future_tx[tx_hash] - self.get_local_height()
if num_blocks_remainining > 0:
return TxMinedInfo(height=TX_HEIGHT_FUTURE, conf=-num_blocks_remainining)
else:
return TxMinedInfo(height=TX_HEIGHT_LOCAL, conf=0)
2018-07-18 11:18:57 +02:00
else:
# local transaction
return TxMinedInfo(height=TX_HEIGHT_LOCAL, conf=0)
2018-07-18 11:18:57 +02:00
2022-12-21 15:23:11 +00:00
def up_to_date_changed(self) -> None:
# fire triggers
util.trigger_callback('adb_set_up_to_date', self)
2018-07-18 11:18:57 +02:00
def is_up_to_date(self):
2022-12-21 15:23:11 +00:00
if not self.synchronizer or not self.verifier:
return False
return self.synchronizer.is_up_to_date() and self.verifier.is_up_to_date()
def reset_netrequest_counters(self) -> None:
if self.synchronizer:
self.synchronizer.reset_request_counters()
if self.verifier:
self.verifier.reset_request_counters()
2019-05-07 17:00:48 +02:00
def get_history_sync_state_details(self) -> Tuple[int, int]:
nsent, nans = 0, 0
2019-05-07 17:00:48 +02:00
if self.synchronizer:
n1, n2 = self.synchronizer.num_requests_sent_and_answered()
nsent += n1
nans += n2
if self.verifier:
n1, n2 = self.verifier.num_requests_sent_and_answered()
nsent += n1
nans += n2
return nsent, nans
2019-05-07 17:00:48 +02:00
@with_transaction_lock
2020-10-18 22:14:52 +02:00
def get_tx_delta(self, tx_hash: str, address: str) -> int:
"""effect of tx on address"""
delta = 0
2020-10-17 17:58:22 +02:00
# subtract the value of coins sent from address
d = self.db.get_txi_addr(tx_hash, address)
for n, v in d:
delta -= v
# add the value of the coins received at address
d = self.db.get_txo_addr(tx_hash, address)
for n, (v, cb) in d.items():
delta += v
return delta
2019-09-12 12:26:49 +02:00
def get_tx_fee(self, txid: str) -> Optional[int]:
""" Returns tx_fee or None. Use server fee only if tx is unconfirmed and not mine"""
# check if stored fee is available
2019-09-12 12:26:49 +02:00
fee = self.db.get_tx_fee(txid, trust_server=False)
if fee is not None:
return fee
# delete server-sent fee for confirmed txns
confirmed = self.get_tx_height(txid).conf > 0
if confirmed:
self.db.add_tx_fee_from_server(txid, None)
# if all inputs are ismine, try to calc fee now;
# otherwise, return stored value
num_all_inputs = self.db.get_num_all_inputs_of_tx(txid)
if num_all_inputs is not None:
2019-09-12 12:26:49 +02:00
# check if tx is mine
num_ismine_inputs = self.db.get_num_ismine_inputs_of_tx(txid)
assert num_ismine_inputs <= num_all_inputs, (num_ismine_inputs, num_all_inputs)
2019-09-12 12:26:49 +02:00
# trust server if tx is unconfirmed and not mine
if num_ismine_inputs < num_all_inputs:
2019-09-12 12:26:49 +02:00
return None if confirmed else self.db.get_tx_fee(txid, trust_server=True)
# lookup tx and deserialize it.
# note that deserializing is expensive, hence above hacks
tx = self.db.get_transaction(txid)
2018-11-18 16:46:07 +01:00
if not tx:
2019-09-12 12:26:49 +02:00
return None
# compute fee if possible
v_in = v_out = 0
with self.lock, self.transaction_lock:
for txin in tx.inputs():
addr = self.get_txin_address(txin)
value = self.get_txin_value(txin, address=addr)
if value is None:
v_in = None
elif v_in is not None:
v_in += value
for txout in tx.outputs():
v_out += txout.value
if v_in is not None:
fee = v_in - v_out
else:
fee = None
# save result
self.db.add_tx_fee_we_calculated(txid, fee)
self.db.add_num_inputs_to_tx(txid, len(tx.inputs()))
2019-09-12 12:26:49 +02:00
return fee
2018-11-18 16:46:07 +01:00
def get_addr_io(self, address):
with self.lock, self.transaction_lock:
h = self.get_address_history(address)
received = {}
sent = {}
for tx_hash, height in h:
hh, pos = self.get_txpos(tx_hash)
d = self.db.get_txo_addr(tx_hash, address)
for n, (v, is_cb) in d.items():
received[tx_hash + ':%d'%n] = (height, pos, v, is_cb)
for tx_hash, height in h:
l = self.db.get_txi_addr(tx_hash, address)
for txi, v in l:
2022-03-19 09:36:50 +01:00
sent[txi] = tx_hash, height
return received, sent
def get_addr_outputs(self, address: str) -> Dict[TxOutpoint, PartialTxInput]:
received, sent = self.get_addr_io(address)
out = {}
for prevout_str, v in received.items():
tx_height, tx_pos, value, is_cb = v
2019-10-23 17:09:41 +02:00
prevout = TxOutpoint.from_str(prevout_str)
utxo = PartialTxInput(prevout=prevout, is_coinbase_output=is_cb)
2019-10-23 17:09:41 +02:00
utxo._trusted_address = address
utxo._trusted_value_sats = value
utxo.block_height = tx_height
utxo.block_txpos = tx_pos
if prevout_str in sent:
txid, height = sent[prevout_str]
2022-03-19 09:36:50 +01:00
utxo.spent_txid = txid
utxo.spent_height = height
else:
utxo.spent_txid = None
utxo.spent_height = None
2019-10-23 17:09:41 +02:00
out[prevout] = utxo
return out
def get_addr_utxo(self, address: str) -> Dict[TxOutpoint, PartialTxInput]:
out = self.get_addr_outputs(address)
for k, v in list(out.items()):
if v.spent_height is not None:
out.pop(k)
return out
# return the total amount ever received by an address
def get_addr_received(self, address):
received, sent = self.get_addr_io(address)
return sum([value for height, pos, value, is_cb in received.values()])
@with_local_height_cached
def get_balance(self, domain, *, excluded_addresses: Set[str] = None,
2022-04-12 09:53:30 +02:00
excluded_coins: Set[str] = None) -> Tuple[int, int, int]:
"""Return the balance of a set of addresses:
confirmed and matured, unconfirmed, unmatured
"""
2022-04-12 09:53:30 +02:00
if excluded_addresses is None:
excluded_addresses = set()
assert isinstance(excluded_addresses, set), f"excluded_addresses should be set, not {type(excluded_addresses)}"
domain = set(domain) - excluded_addresses
2019-02-24 08:41:00 +01:00
if excluded_coins is None:
excluded_coins = set()
assert isinstance(excluded_coins, set), f"excluded_coins should be set, not {type(excluded_coins)}"
2022-04-12 09:53:30 +02:00
cache_key = sha256(','.join(sorted(domain)) + ';'
+ ','.join(sorted(excluded_coins)))
2022-04-12 09:53:30 +02:00
cached_value = self._get_balance_cache.get(cache_key)
if cached_value:
return cached_value
coins = {}
for address in domain:
coins.update(self.get_addr_outputs(address))
c = u = x = 0
2019-09-08 18:26:04 +02:00
mempool_height = self.get_local_height() + 1 # height of next block
for utxo in coins.values(): # type: PartialTxInput
2022-04-12 09:53:30 +02:00
if utxo.spent_height is not None:
2019-02-24 08:41:00 +01:00
continue
2022-04-12 09:53:30 +02:00
if utxo.prevout.to_str() in excluded_coins:
continue
v = utxo.value_sats()
tx_height = utxo.block_height
is_cb = utxo.is_coinbase_output()
2019-09-08 18:26:04 +02:00
if is_cb and tx_height + COINBASE_MATURITY > mempool_height:
x += v
elif tx_height > 0:
c += v
else:
2022-04-12 09:53:30 +02:00
txid = utxo.prevout.txid.hex()
tx = self.db.get_transaction(txid)
assert tx is not None # txid comes from get_addr_io
# we look at the outputs that are spent by this transaction
# if those outputs are ours and confirmed, we count this coin as confirmed
confirmed_spent_amount = 0
for txin in tx.inputs():
if txin.prevout in coins:
coin = coins[txin.prevout]
if coin.block_height > 0:
confirmed_spent_amount += coin.value_sats()
# Compare amount, in case tx has confirmed and unconfirmed inputs, or is a coinjoin.
# (fixme: tx may have multiple change outputs)
if confirmed_spent_amount >= v:
c += v
else:
2022-04-12 09:53:30 +02:00
c += confirmed_spent_amount
u += v - confirmed_spent_amount
2019-03-04 22:19:33 +01:00
result = c, u, x
# cache result.
2022-04-12 09:53:30 +02:00
# Cache needs to be invalidated if a transaction is added to/
# removed from history; or on new blocks (maturity...)
self._get_balance_cache[cache_key] = result
2019-03-04 22:19:33 +01:00
return result
@with_local_height_cached
def get_utxos(
self,
domain,
*,
excluded_addresses=None,
mature_only: bool = False,
confirmed_funding_only: bool = False,
confirmed_spending_only: bool = False,
nonlocal_only: bool = False,
block_height: int = None,
) -> Sequence[PartialTxInput]:
if block_height is not None:
# caller wants the UTXOs we had at a given height; check other parameters
assert confirmed_funding_only
assert confirmed_spending_only
assert nonlocal_only
else:
block_height = self.get_local_height()
coins = []
domain = set(domain)
2019-02-24 08:41:00 +01:00
if excluded_addresses:
domain = set(domain) - set(excluded_addresses)
mempool_height = block_height + 1 # height of next block
for addr in domain:
txos = self.get_addr_outputs(addr)
for txo in txos.values():
if txo.spent_height is not None:
if not confirmed_spending_only:
continue
if confirmed_spending_only and 0 < txo.spent_height <= block_height:
continue
if confirmed_funding_only and not (0 < txo.block_height <= block_height):
continue
if nonlocal_only and txo.block_height in (TX_HEIGHT_LOCAL, TX_HEIGHT_FUTURE):
continue
if (mature_only and txo.is_coinbase_output()
and txo.block_height + COINBASE_MATURITY > mempool_height):
continue
coins.append(txo)
continue
return coins
def is_used(self, address: str) -> bool:
return self.get_address_history_len(address) != 0
def is_empty(self, address: str) -> bool:
2022-04-12 09:53:30 +02:00
coins = self.get_addr_utxo(address)
return not bool(coins)
2022-06-07 16:52:09 +02:00
@with_local_height_cached
def address_is_old(self, address: str, *, req_conf: int = 3) -> bool:
"""Returns whether address has any history that is deeply confirmed.
Used for reorg-safe(ish) gap limit roll-forward.
"""
max_conf = -1
h = self.db.get_addr_history(address)
needs_spv_check = not self.config.get("skipmerklecheck", False)
for tx_hash, tx_height in h:
if needs_spv_check:
tx_age = self.get_tx_height(tx_hash).conf
else:
if tx_height <= 0:
tx_age = 0
else:
tx_age = self.get_local_height() - tx_height + 1
max_conf = max(max_conf, tx_age)
return max_conf >= req_conf