LBRYFoundation/LBRY-Vault
1
0
Fork 0
mirror of https://github.com/LBRYFoundation/LBRY-Vault.git synced 2025-09-05 05:15:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

abstract database away from wallet and address_synchronizer

Browse source
This commit is contained in:
ThomasV 2019-02-20 14:45:15 +01:00
parent 7f2083f667
commit 791e680a96
6 changed files with 258 additions and 189 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

211
electrum/address_synchronizer.py
View file

@ -61,6 +61,7 @@ class AddressSynchronizer(PrintError):
def __init__(self, storage: 'WalletStorage'): def __init__(self, storage: 'WalletStorage'):
self.storage = storage self.storage = storage
self.db = self.storage.db
self.network = None # type: Network self.network = None # type: Network
# verifier (SPV) and synchronizer are started in start_network # verifier (SPV) and synchronizer are started in start_network
self.synchronizer = None # type: Synchronizer self.synchronizer = None # type: Synchronizer
@ -68,17 +69,6 @@ class AddressSynchronizer(PrintError):
# locks: if you need to take multiple ones, acquire them in the order they are defined here! # locks: if you need to take multiple ones, acquire them in the order they are defined here!
self.lock = threading.RLock() self.lock = threading.RLock()
self.transaction_lock = threading.RLock() self.transaction_lock = threading.RLock()
# address -> list(txid, height)
self.history = storage.get('addr_history',{})
# Verified transactions. txid -> TxMinedInfo. Access with self.lock.
verified_tx = storage.get('verified_tx3', {})
self.verified_tx = {} # type: Dict[str, TxMinedInfo]
for txid, (height, timestamp, txpos, header_hash) in verified_tx.items():
self.verified_tx[txid] = TxMinedInfo(height=height,
conf=None,
timestamp=timestamp,
txpos=txpos,
header_hash=header_hash)
# Transactions pending verification. txid -> tx_height. Access with self.lock. # Transactions pending verification. txid -> tx_height. Access with self.lock.
self.unverified_tx = defaultdict(int) self.unverified_tx = defaultdict(int)
# true when synchronized # true when synchronized
@ -95,17 +85,16 @@ class AddressSynchronizer(PrintError):
return func_wrapper return func_wrapper
def load_and_cleanup(self): def load_and_cleanup(self):
self.load_transactions()
self.load_local_history() self.load_local_history()
self.check_history() self.check_history()
self.load_unverified_transactions() self.load_unverified_transactions()
self.remove_local_transactions_we_dont_have() self.remove_local_transactions_we_dont_have()
def is_mine(self, address): def is_mine(self, address):
return address in self.history return address in self.db.get_history()
def get_addresses(self): def get_addresses(self):
return sorted(self.history.keys()) return sorted(self.db.get_history())
def get_address_history(self, addr): def get_address_history(self, addr):
h = [] h = []
@ -128,8 +117,8 @@ class AddressSynchronizer(PrintError):
return addr return addr
prevout_hash = txi.get('prevout_hash') prevout_hash = txi.get('prevout_hash')
prevout_n = txi.get('prevout_n') prevout_n = txi.get('prevout_n')
dd = self.txo.get(prevout_hash, {}) for addr in self.db.get_txo(prevout_hash):
for addr, l in dd.items(): l = self.db.get_txo_addr(prevout_hash, addr)
for n, v, is_cb in l: for n, v, is_cb in l:
if n == prevout_n: if n == prevout_n:
return addr return addr
@ -146,7 +135,8 @@ class AddressSynchronizer(PrintError):
def load_unverified_transactions(self): def load_unverified_transactions(self):
# review transactions that are in the history # review transactions that are in the history
for addr, hist in self.history.items(): for addr in self.db.get_history():
hist = self.db.get_addr_history(addr)
for tx_hash, tx_height in hist: for tx_hash, tx_height in hist:
# add it in case it was previously unconfirmed # add it in case it was previously unconfirmed
self.add_unverified_tx(tx_hash, tx_height) self.add_unverified_tx(tx_hash, tx_height)
@ -167,13 +157,11 @@ class AddressSynchronizer(PrintError):
self.verifier = None self.verifier = None
self.storage.put('stored_height', self.get_local_height()) self.storage.put('stored_height', self.get_local_height())
if write_to_disk: if write_to_disk:
self.save_transactions()
self.save_verified_tx()
self.storage.write() self.storage.write()
def add_address(self, address): def add_address(self, address):
if address not in self.history: if address not in self.db.get_history():
self.history[address] = [] self.db.history[address] = []
self.set_up_to_date(False) self.set_up_to_date(False)
if self.synchronizer: if self.synchronizer:
self.synchronizer.add(address) self.synchronizer.add(address)
@ -191,11 +179,11 @@ class AddressSynchronizer(PrintError):
continue continue
prevout_hash = txin['prevout_hash'] prevout_hash = txin['prevout_hash']
prevout_n = txin['prevout_n'] prevout_n = txin['prevout_n']
spending_tx_hash = self.spent_outpoints[prevout_hash].get(prevout_n) spending_tx_hash = self.db.get_spent_outpoint(prevout_hash, prevout_n)
if spending_tx_hash is None: if spending_tx_hash is None:
continue continue
# this outpoint has already been spent, by spending_tx # this outpoint has already been spent, by spending_tx
assert spending_tx_hash in self.transactions assert spending_tx_hash in self.db.list_transactions()
conflicting_txns |= {spending_tx_hash} conflicting_txns |= {spending_tx_hash}
if tx_hash in conflicting_txns: if tx_hash in conflicting_txns:
# this tx is already in history, so it conflicts with itself # this tx is already in history, so it conflicts with itself
@ -256,49 +244,39 @@ class AddressSynchronizer(PrintError):
self.remove_transaction(tx_hash2) self.remove_transaction(tx_hash2)
# add inputs # add inputs
def add_value_from_prev_output(): def add_value_from_prev_output():
dd = self.txo.get(prevout_hash, {})
# note: this nested loop takes linear time in num is_mine outputs of prev_tx # note: this nested loop takes linear time in num is_mine outputs of prev_tx
for addr, outputs in dd.items(): for addr in self.db.get_txo(prevout_hash):
outputs = self.db.get_txo_addr(prevout_hash, addr)
# note: instead of [(n, v, is_cb), ...]; we could store: {n -> (v, is_cb)} # note: instead of [(n, v, is_cb), ...]; we could store: {n -> (v, is_cb)}
for n, v, is_cb in outputs: for n, v, is_cb in outputs:
if n == prevout_n: if n == prevout_n:
if addr and self.is_mine(addr): if addr and self.is_mine(addr):
if d.get(addr) is None: self.db.add_txi_addr(tx_hash, addr, ser, v)
d[addr] = set()
d[addr].add((ser, v))
return return
self.txi[tx_hash] = d = {}
for txi in tx.inputs(): for txi in tx.inputs():
if txi['type'] == 'coinbase': if txi['type'] == 'coinbase':
continue continue
prevout_hash = txi['prevout_hash'] prevout_hash = txi['prevout_hash']
prevout_n = txi['prevout_n'] prevout_n = txi['prevout_n']
ser = prevout_hash + ':%d' % prevout_n ser = prevout_hash + ':%d' % prevout_n
self.spent_outpoints[prevout_hash][prevout_n] = tx_hash self.db.set_spent_outpoint(prevout_hash, prevout_n, tx_hash)
add_value_from_prev_output() add_value_from_prev_output()
# add outputs # add outputs
self.txo[tx_hash] = d = {}
for n, txo in enumerate(tx.outputs()): for n, txo in enumerate(tx.outputs()):
v = txo[2] v = txo[2]
ser = tx_hash + ':%d'%n ser = tx_hash + ':%d'%n
addr = self.get_txout_address(txo) addr = self.get_txout_address(txo)
if addr and self.is_mine(addr): if addr and self.is_mine(addr):
if d.get(addr) is None: self.db.add_txo_addr(tx_hash, addr, n, v, is_coinbase)
d[addr] = []
d[addr].append((n, v, is_coinbase))
# give v to txi that spends me # give v to txi that spends me
next_tx = self.spent_outpoints[tx_hash].get(n) next_tx = self.db.get_spent_outpoint(tx_hash, n)
if next_tx is not None: if next_tx is not None:
dd = self.txi.get(next_tx, {}) self.db.add_txi_addr(next_tx, addr, ser, v)
if dd.get(addr) is None:
dd[addr] = set()
if (ser, v) not in dd[addr]:
dd[addr].add((ser, v))
self._add_tx_to_local_history(next_tx) self._add_tx_to_local_history(next_tx)
# add to local history # add to local history
self._add_tx_to_local_history(tx_hash) self._add_tx_to_local_history(tx_hash)
# save # save
self.transactions[tx_hash] = tx self.db.add_transaction(tx_hash, tx)
return True return True
def remove_transaction(self, tx_hash): def remove_transaction(self, tx_hash):
@ -328,17 +306,18 @@ class AddressSynchronizer(PrintError):
with self.transaction_lock: with self.transaction_lock:
self.print_error("removing tx from history", tx_hash) self.print_error("removing tx from history", tx_hash)
tx = self.transactions.pop(tx_hash, None) tx = self.db.remove_transaction(tx_hash)
remove_from_spent_outpoints() remove_from_spent_outpoints()
self._remove_tx_from_local_history(tx_hash) self._remove_tx_from_local_history(tx_hash)
self.txi.pop(tx_hash, None) self.db.remove_txi(tx_hash)
self.txo.pop(tx_hash, None) self.db.remove_txo(tx_hash)
def get_depending_transactions(self, tx_hash): def get_depending_transactions(self, tx_hash):
"""Returns all (grand-)children of tx_hash in this wallet.""" """Returns all (grand-)children of tx_hash in this wallet."""
with self.transaction_lock: with self.transaction_lock:
children = set() children = set()
for other_hash in self.spent_outpoints[tx_hash].values(): for n in self.db.get_spent_outpoints(tx_hash):
other_hash = self.db.get_spent_outpoint(tx_hash, n)
children.add(other_hash) children.add(other_hash)
children |= self.get_depending_transactions(other_hash) children |= self.get_depending_transactions(other_hash)
return children return children
@ -354,129 +333,68 @@ class AddressSynchronizer(PrintError):
if (tx_hash, height) not in hist: if (tx_hash, height) not in hist:
# make tx local # make tx local
self.unverified_tx.pop(tx_hash, None) self.unverified_tx.pop(tx_hash, None)
self.verified_tx.pop(tx_hash, None) self.db.remove_verified_tx(tx_hash)
if self.verifier: if self.verifier:
self.verifier.remove_spv_proof_for_tx(tx_hash) self.verifier.remove_spv_proof_for_tx(tx_hash)
self.history[addr] = hist self.db.set_addr_history(addr, hist)
for tx_hash, tx_height in hist: for tx_hash, tx_height in hist:
# add it in case it was previously unconfirmed # add it in case it was previously unconfirmed
self.add_unverified_tx(tx_hash, tx_height) self.add_unverified_tx(tx_hash, tx_height)
# if addr is new, we have to recompute txi and txo # if addr is new, we have to recompute txi and txo
tx = self.transactions.get(tx_hash) tx = self.db.get_transaction(tx_hash)
if tx is None: if tx is None:
continue continue
self.add_transaction(tx_hash, tx, allow_unrelated=True) self.add_transaction(tx_hash, tx, allow_unrelated=True)
# Store fees # Store fees
self.tx_fees.update(tx_fees) self.db.update_tx_fees(tx_fees)
@profiler
def load_transactions(self):
# load txi, txo, tx_fees
# bookkeeping data of is_mine inputs of transactions
self.txi = self.storage.get('txi', {}) # txid -> address -> (prev_outpoint, value)
for txid, d in list(self.txi.items()):
for addr, lst in d.items():
self.txi[txid][addr] = set([tuple(x) for x in lst])
# bookkeeping data of is_mine outputs of transactions
self.txo = self.storage.get('txo', {}) # txid -> address -> (output_index, value, is_coinbase)
self.tx_fees = self.storage.get('tx_fees', {})
tx_list = self.storage.get('transactions', {})
# load transactions
self.transactions = {}
for tx_hash, raw in tx_list.items():
tx = Transaction(raw)
self.transactions[tx_hash] = tx
if self.txi.get(tx_hash) is None and self.txo.get(tx_hash) is None:
self.print_error("removing unreferenced tx", tx_hash)
self.transactions.pop(tx_hash)
# load spent_outpoints
_spent_outpoints = self.storage.get('spent_outpoints', {})
self.spent_outpoints = defaultdict(dict)
for prevout_hash, d in _spent_outpoints.items():
for prevout_n_str, spending_txid in d.items():
prevout_n = int(prevout_n_str)
if spending_txid not in self.transactions:
continue # only care about txns we have
self.spent_outpoints[prevout_hash][prevout_n] = spending_txid
@profiler @profiler
def load_local_history(self): def load_local_history(self):
self._history_local = {} # address -> set(txid) self._history_local = {} # address -> set(txid)
self._address_history_changed_events = defaultdict(asyncio.Event) # address -> Event self._address_history_changed_events = defaultdict(asyncio.Event) # address -> Event
for txid in itertools.chain(self.txi, self.txo): for txid in itertools.chain(self.db.get_txi_keys(), self.db.get_txo_keys()):
self._add_tx_to_local_history(txid) self._add_tx_to_local_history(txid)
@profiler @profiler
def check_history(self): def check_history(self):
save = False save = False
hist_addrs_mine = list(filter(lambda k: self.is_mine(k), self.history.keys())) 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.history.keys())) hist_addrs_not_mine = list(filter(lambda k: not self.is_mine(k), self.db.get_history()))
for addr in hist_addrs_not_mine: for addr in hist_addrs_not_mine:
self.history.pop(addr) self.db.remove_addr_history(addr)
save = True save = True
for addr in hist_addrs_mine: for addr in hist_addrs_mine:
hist = self.history[addr] hist = self.db.get_addr_history(addr)
for tx_hash, tx_height in hist: for tx_hash, tx_height in hist:
if self.txi.get(tx_hash) or self.txo.get(tx_hash): if self.db.get_txi(tx_hash) or self.db.get_txo(tx_hash):
continue continue
tx = self.transactions.get(tx_hash) tx = self.db.get_transaction(tx_hash)
if tx is not None: if tx is not None:
self.add_transaction(tx_hash, tx, allow_unrelated=True) self.add_transaction(tx_hash, tx, allow_unrelated=True)
save = True save = True
if save: if save:
self.save_transactions() self.storage.write()
def remove_local_transactions_we_dont_have(self): def remove_local_transactions_we_dont_have(self):
txid_set = set(self.txi) | set(self.txo) for txid in itertools.chain(list(self.db.get_txi_keys()), list(self.db.get_txo_keys())):
for txid in txid_set:
tx_height = self.get_tx_height(txid).height tx_height = self.get_tx_height(txid).height
if tx_height == TX_HEIGHT_LOCAL and txid not in self.transactions: if tx_height == TX_HEIGHT_LOCAL and txid not in self.db.list_transactions():
self.remove_transaction(txid) self.remove_transaction(txid)
@profiler
def save_transactions(self, write=False):
with self.transaction_lock:
tx = {}
for k,v in self.transactions.items():
tx[k] = str(v)
self.storage.put('transactions', tx)
self.storage.put('txi', self.txi)
self.storage.put('txo', self.txo)
self.storage.put('tx_fees', self.tx_fees)
self.storage.put('addr_history', self.history)
self.storage.put('spent_outpoints', self.spent_outpoints)
if write:
self.storage.write()
def save_verified_tx(self, write=False):
with self.lock:
verified_tx_to_save = {}
for txid, tx_info in self.verified_tx.items():
verified_tx_to_save[txid] = (tx_info.height, tx_info.timestamp,
tx_info.txpos, tx_info.header_hash)
self.storage.put('verified_tx3', verified_tx_to_save)
if write:
self.storage.write()
def clear_history(self): def clear_history(self):
with self.lock: with self.lock:
with self.transaction_lock: with self.transaction_lock:
self.txi = {} self.db.clear_history()
self.txo = {} self.storage.write()
self.tx_fees = {}
self.spent_outpoints = defaultdict(dict)
self.history = {}
self.verified_tx = {}
self.transactions = {} # type: Dict[str, Transaction]
self.save_transactions()
def get_txpos(self, tx_hash): def get_txpos(self, tx_hash):
"""Returns (height, txpos) tuple, even if the tx is unverified.""" """Returns (height, txpos) tuple, even if the tx is unverified."""
with self.lock: with self.lock:
if tx_hash in self.verified_tx: if tx_hash in self.db.list_verified_tx():
info = self.verified_tx[tx_hash] info = self.db.get_verified_tx(tx_hash)
return info.height, info.txpos return info.height, info.txpos
elif tx_hash in self.unverified_tx: elif tx_hash in self.unverified_tx:
height = self.unverified_tx[tx_hash] height = self.unverified_tx[tx_hash]
@ -540,7 +458,7 @@ class AddressSynchronizer(PrintError):
def _add_tx_to_local_history(self, txid): def _add_tx_to_local_history(self, txid):
with self.transaction_lock: with self.transaction_lock:
for addr in itertools.chain(self.txi.get(txid, []), self.txo.get(txid, [])): for addr in itertools.chain(self.db.get_txi(txid), self.db.get_txo(txid)):
cur_hist = self._history_local.get(addr, set()) cur_hist = self._history_local.get(addr, set())
cur_hist.add(txid) cur_hist.add(txid)
self._history_local[addr] = cur_hist self._history_local[addr] = cur_hist
@ -548,7 +466,7 @@ class AddressSynchronizer(PrintError):
def _remove_tx_from_local_history(self, txid): def _remove_tx_from_local_history(self, txid):
with self.transaction_lock: with self.transaction_lock:
for addr in itertools.chain(self.txi.get(txid, []), self.txo.get(txid, [])): for addr in itertools.chain(self.db.get_txi(txid), self.db.get_txo(txid)):
cur_hist = self._history_local.get(addr, set()) cur_hist = self._history_local.get(addr, set())
try: try:
cur_hist.remove(txid) cur_hist.remove(txid)
@ -573,10 +491,10 @@ class AddressSynchronizer(PrintError):
await self._address_history_changed_events[addr].wait() await self._address_history_changed_events[addr].wait()
def add_unverified_tx(self, tx_hash, tx_height): def add_unverified_tx(self, tx_hash, tx_height):
if tx_hash in self.verified_tx: if tx_hash in self.db.list_verified_tx():
if tx_height in (TX_HEIGHT_UNCONFIRMED, TX_HEIGHT_UNCONF_PARENT): if tx_height in (TX_HEIGHT_UNCONFIRMED, TX_HEIGHT_UNCONF_PARENT):
with self.lock: with self.lock:
self.verified_tx.pop(tx_hash) self.db.remove_verified_tx(tx_hash)
if self.verifier: if self.verifier:
self.verifier.remove_spv_proof_for_tx(tx_hash) self.verifier.remove_spv_proof_for_tx(tx_hash)
else: else:
@ -594,7 +512,7 @@ class AddressSynchronizer(PrintError):
# Remove from the unverified map and add to the verified map # Remove from the unverified map and add to the verified map
with self.lock: with self.lock:
self.unverified_tx.pop(tx_hash, None) self.unverified_tx.pop(tx_hash, None)
self.verified_tx[tx_hash] = info self.db.add_verified_tx(tx_hash, info)
tx_mined_status = self.get_tx_height(tx_hash) tx_mined_status = self.get_tx_height(tx_hash)
self.network.trigger_callback('verified', self, tx_hash, tx_mined_status) self.network.trigger_callback('verified', self, tx_hash, tx_mined_status)
@ -607,12 +525,13 @@ class AddressSynchronizer(PrintError):
'''Used by the verifier when a reorg has happened''' '''Used by the verifier when a reorg has happened'''
txs = set() txs = set()
with self.lock: with self.lock:
for tx_hash, info in list(self.verified_tx.items()): for tx_hash in list(self.db.list_verified_tx()):
info = self.db.get_verified_tx(tx_hash)
tx_height = info.height tx_height = info.height
if tx_height >= height: if tx_height >= height:
header = blockchain.read_header(tx_height) header = blockchain.read_header(tx_height)
if not header or hash_header(header) != info.header_hash: if not header or hash_header(header) != info.header_hash:
self.verified_tx.pop(tx_hash, None) self.db.remove_verified_tx(tx_hash)
# NOTE: we should add these txns to self.unverified_tx, # NOTE: we should add these txns to self.unverified_tx,
# but with what height? # but with what height?
# If on the new fork after the reorg, the txn is at the # If on the new fork after the reorg, the txn is at the
@ -635,8 +554,8 @@ class AddressSynchronizer(PrintError):
def get_tx_height(self, tx_hash: str) -> TxMinedInfo: def get_tx_height(self, tx_hash: str) -> TxMinedInfo:
with self.lock: with self.lock:
if tx_hash in self.verified_tx: if tx_hash in self.db.list_verified_tx():
info = self.verified_tx[tx_hash] info = self.db.get_verified_tx(tx_hash)
conf = max(self.get_local_height() - info.height + 1, 0) conf = max(self.get_local_height() - info.height + 1, 0)
return info._replace(conf=conf) return info._replace(conf=conf)
elif tx_hash in self.unverified_tx: elif tx_hash in self.unverified_tx:
@ -652,11 +571,7 @@ class AddressSynchronizer(PrintError):
if self.network: if self.network:
self.network.notify('status') self.network.notify('status')
if up_to_date: if up_to_date:
self.save_transactions(write=True) self.storage.write()
# if the verifier is also up to date, persist that too;
# otherwise it will persist its results when it finishes
if self.verifier and self.verifier.is_up_to_date():
self.save_verified_tx(write=True)
def is_up_to_date(self): def is_up_to_date(self):
with self.lock: return self.up_to_date with self.lock: return self.up_to_date
@ -666,11 +581,11 @@ class AddressSynchronizer(PrintError):
"""effect of tx on address""" """effect of tx on address"""
delta = 0 delta = 0
# substract the value of coins sent from address # substract the value of coins sent from address
d = self.txi.get(tx_hash, {}).get(address, []) d = self.db.get_txi_addr(tx_hash, address)
for n, v in d: for n, v in d:
delta -= v delta -= v
# add the value of the coins received at address # add the value of the coins received at address
d = self.txo.get(tx_hash, {}).get(address, []) d = self.db.get_txo_addr(tx_hash, address)
for n, v, cb in d: for n, v, cb in d:
delta += v delta += v
return delta return delta
@ -679,10 +594,12 @@ class AddressSynchronizer(PrintError):
def get_tx_value(self, txid): def get_tx_value(self, txid):
"""effect of tx on the entire domain""" """effect of tx on the entire domain"""
delta = 0 delta = 0
for addr, d in self.txi.get(txid, {}).items(): for addr in self.db.get_txi(txid):
d = self.db.get_txi_addr(txid, addr)
for n, v in d: for n, v in d:
delta -= v delta -= v
for addr, d in self.txo.get(txid, {}).items(): for addr in self.db.get_txo(txid):
d = self.db.get_txo_addr(txid, addr)
for n, v, cb in d: for n, v, cb in d:
delta += v delta += v
return delta return delta
@ -699,7 +616,7 @@ class AddressSynchronizer(PrintError):
if self.is_mine(addr): if self.is_mine(addr):
is_mine = True is_mine = True
is_relevant = True is_relevant = True
d = self.txo.get(txin['prevout_hash'], {}).get(addr, []) d = self.db.get_txo_addr(txin['prevout_hash'], addr)
for n, v, cb in d: for n, v, cb in d:
if n == txin['prevout_n']: if n == txin['prevout_n']:
value = v value = v
@ -748,7 +665,7 @@ class AddressSynchronizer(PrintError):
is_relevant, is_mine, v, fee = self.get_wallet_delta(tx) is_relevant, is_mine, v, fee = self.get_wallet_delta(tx)
if fee is None: if fee is None:
txid = tx.txid() txid = tx.txid()
fee = self.tx_fees.get(txid) fee = self.db.get_tx_fee(txid)
# only cache non-None, as None can still change while syncing # only cache non-None, as None can still change while syncing
if fee is not None: if fee is not None:
tx._cached_fee = fee tx._cached_fee = fee
@ -760,11 +677,11 @@ class AddressSynchronizer(PrintError):
received = {} received = {}
sent = {} sent = {}
for tx_hash, height in h: for tx_hash, height in h:
l = self.txo.get(tx_hash, {}).get(address, []) l = self.db.get_txo_addr(tx_hash, address)
for n, v, is_cb in l: for n, v, is_cb in l:
received[tx_hash + ':%d'%n] = (height, v, is_cb) received[tx_hash + ':%d'%n] = (height, v, is_cb)
for tx_hash, height in h: for tx_hash, height in h:
l = self.txi.get(tx_hash, {}).get(address, []) l = self.db.get_txi_addr(tx_hash, address)
for txi, v in l: for txi, v in l:
sent[txi] = height sent[txi] = height
return received, sent return received, sent
@ -849,7 +766,7 @@ class AddressSynchronizer(PrintError):
return cc, uu, xx return cc, uu, xx
def is_used(self, address): def is_used(self, address):
h = self.history.get(address,[]) h = self.db.get_addr_history(address)
return len(h) != 0 return len(h) != 0
def is_empty(self, address): def is_empty(self, address):

156
electrum/json_db.py
View file

@ -35,10 +35,10 @@ import zlib
from collections import defaultdict from collections import defaultdict
from . import util, bitcoin, ecc from . import util, bitcoin, ecc
from .util import PrintError, profiler, InvalidPassword, WalletFileException, bfh, standardize_path, multisig_type from .util import PrintError, profiler, InvalidPassword, WalletFileException, bfh, standardize_path, multisig_type, TxMinedInfo
from .plugin import run_hook, plugin_loaders from .plugin import run_hook, plugin_loaders
from .keystore import bip44_derivation from .keystore import bip44_derivation
from .transaction import Transaction
# seed_version is now used for the version of the wallet file # seed_version is now used for the version of the wallet file
@ -49,9 +49,6 @@ FINAL_SEED_VERSION = 18 # electrum >= 2.7 will set this to prevent
class JsonDB(PrintError): class JsonDB(PrintError):
def __init__(self, raw, manual_upgrades): def __init__(self, raw, manual_upgrades):
@ -61,6 +58,7 @@ class JsonDB(PrintError):
self.load_data(raw) self.load_data(raw)
else: else:
self.put('seed_version', FINAL_SEED_VERSION) self.put('seed_version', FINAL_SEED_VERSION)
self.load_transactions()
def get(self, key, default=None): def get(self, key, default=None):
v = self.data.get(key) v = self.data.get(key)
@ -498,3 +496,151 @@ class JsonDB(PrintError):
msg += "\nPlease open this file with Electrum 1.9.8, and move your coins to a new wallet." msg += "\nPlease open this file with Electrum 1.9.8, and move your coins to a new wallet."
raise WalletFileException(msg) raise WalletFileException(msg)
def get_txi(self, tx_hash):
return self.txi.get(tx_hash, {}).keys()
def get_txo(self, tx_hash):
return self.txo.get(tx_hash, {}).keys()
def get_txi_addr(self, tx_hash, address):
return self.txi.get(tx_hash, {}).get(address, [])
def get_txo_addr(self, tx_hash, address):
return self.txo.get(tx_hash, {}).get(address, [])
def add_txi_addr(self, tx_hash, addr, ser, v):
if tx_hash not in self.txi:
self.txi[tx_hash] = {}
d = self.txi[tx_hash]
if addr not in d:
d[addr] = set()
d[addr].add((ser, v))
def add_txo_addr(self, tx_hash, addr, n, v, is_coinbase):
if tx_hash not in self.txo:
self.txo[tx_hash] = {}
d = self.txo[tx_hash]
if addr not in d:
d[addr] = []
d[addr].append((n, v, is_coinbase))
def get_txi_keys(self):
return self.txi.keys()
def get_txo_keys(self):
return self.txo.keys()
def remove_txi(self, tx_hash):
self.txi.pop(tx_hash, None)
def remove_txo(self, tx_hash):
self.txo.pop(tx_hash, None)
def get_spent_outpoints(self, prevout_hash):
return self.spent_outpoints.get(prevout_hash, {}).keys()
def get_spent_outpoint(self, prevout_hash, prevout_n):
return self.spent_outpoints.get(prevout_hash, {}).get(str(prevout_n))
def set_spent_outpoint(self, prevout_hash, prevout_n, tx_hash):
if prevout_hash not in self.spent_outpoints:
self.spent_outpoints[prevout_hash] = {}
self.spent_outpoints[prevout_hash][str(prevout_n)] = tx_hash
def add_transaction(self, tx_hash, tx):
self.transactions[tx_hash] = str(tx)
def remove_transaction(self, tx_hash):
self.transactions.pop(tx_hash, None)
def get_transaction(self, tx_hash):
tx = self.transactions.get(tx_hash)
return Transaction(tx) if tx else None
def list_transactions(self):
return self.transactions.keys()
def get_history(self):
return self.history.keys()
def get_addr_history(self, addr):
return self.history.get(addr, [])
def set_addr_history(self, addr, hist):
self.history[addr] = hist
def remove_addr_history(self, addr):
self.history.pop(addr, None)
def list_verified_tx(self):
return self.verified_tx.keys()
def get_verified_tx(self, txid):
if txid not in self.verified_tx:
return None
height, timestamp, txpos, header_hash = self.verified_tx[txid]
return TxMinedInfo(height=height,
conf=None,
timestamp=timestamp,
txpos=txpos,
header_hash=header_hash)
def add_verified_tx(self, txid, info):
self.verified_tx[txid] = (info.height, info.timestamp, info.txpos, info.header_hash)
def remove_verified_tx(self, txid):
self.verified_tx.pop(txid, None)
def update_tx_fees(self, d):
return self.tx_fees.update(d)
def get_tx_fee(self, txid):
return self.tx_fees.get(txid)
def remove_tx_fee(self, txid):
self.tx_fees.pop(txid, None)
def get_data_ref(self, name):
if name not in self.data:
self.data[name] = {}
return self.data[name]
@profiler
def load_transactions(self):
# references in self.data
self.txi = self.get_data_ref('txi') # txid -> address -> (prev_outpoint, value)
self.txo = self.get_data_ref('txo') # txid -> address -> (output_index, value, is_coinbase)
self.transactions = self.get_data_ref('transactions') # type: Dict[str, Transaction]
self.spent_outpoints = self.get_data_ref('spent_outpoints')
self.history = self.get_data_ref('history') # address -> list(txid, height)
self.verified_tx = self.get_data_ref('verified_tx3') # txid -> TxMinedInfo. Access with self.lock.
self.tx_fees = self.get_data_ref('tx_fees')
# tuple to set
for txid, d in list(self.txi.items()):
for addr, lst in d.items():
self.txi[txid][addr] = set([tuple(x) for x in lst])
# remove unreferenced tx
for tx_hash in self.transactions:
if not self.get_txi(tx_hash) and not self.get_txo(tx_hash):
self.print_error("removing unreferenced tx", tx_hash)
self.transactions.pop(tx_hash)
# remove unreferenced outpoints
for prevout_hash in self.spent_outpoints.keys():
d = self.spent_outpoints[prevout_hash]
for prevout_n, spending_txid in list(d.items()):
if spending_txid not in self.transactions:
self.print_error("removing unreferenced spent outpoint")
d.pop(prevout_n)
def clear_history(self):
self.txi.clear()
self.txo.clear()
self.spent_outpoints.clear()
self.transactions.clear()
self.history.clear()
self.verified_tx.clear()
self.tx_fees.clear()

2
electrum/network.py
View file

@ -1184,8 +1184,6 @@ class Network(PrintError):
coro = asyncio.run_coroutine_threadsafe(cls._send_http_on_proxy(method, url, **kwargs), loop) coro = asyncio.run_coroutine_threadsafe(cls._send_http_on_proxy(method, url, **kwargs), loop)
return coro.result(5) return coro.result(5)
# methods used in scripts # methods used in scripts
async def get_peers(self): async def get_peers(self):
while not self.is_connected(): while not self.is_connected():

7
electrum/synchronizer.py
View file

@ -139,7 +139,7 @@ class Synchronizer(SynchronizerBase):
and not self.requested_tx) and not self.requested_tx)
async def _on_address_status(self, addr, status): async def _on_address_status(self, addr, status):
history = self.wallet.history.get(addr, []) history = self.wallet.db.get_addr_history(addr)
if history_status(history) == status: if history_status(history) == status:
return return
if addr in self.requested_histories: if addr in self.requested_histories:
@ -175,7 +175,7 @@ class Synchronizer(SynchronizerBase):
for tx_hash, tx_height in hist: for tx_hash, tx_height in hist:
if tx_hash in self.requested_tx: if tx_hash in self.requested_tx:
continue continue
if tx_hash in self.wallet.transactions: if tx_hash in self.wallet.db.list_transactions():
continue continue
transaction_hashes.append(tx_hash) transaction_hashes.append(tx_hash)
self.requested_tx[tx_hash] = tx_height self.requested_tx[tx_hash] = tx_height
@ -216,7 +216,8 @@ class Synchronizer(SynchronizerBase):
async def main(self): async def main(self):
self.wallet.set_up_to_date(False) self.wallet.set_up_to_date(False)
# request missing txns, if any # request missing txns, if any
for history in self.wallet.history.values(): for addr in self.wallet.db.get_history():
history = self.wallet.db.get_addr_history(addr)
# Old electrum servers returned ['*'] when all history for the address # Old electrum servers returned ['*'] when all history for the address
# was pruned. This no longer happens but may remain in old wallets. # was pruned. This no longer happens but may remain in old wallets.
if history == ['*']: continue if history == ['*']: continue

4
electrum/verifier.py
View file

@ -135,8 +135,8 @@ class SPV(NetworkJobOnDefaultServer):
txpos=pos, txpos=pos,
header_hash=header_hash) header_hash=header_hash)
self.wallet.add_verified_tx(tx_hash, tx_info) self.wallet.add_verified_tx(tx_hash, tx_info)
if self.is_up_to_date() and self.wallet.is_up_to_date(): #if self.is_up_to_date() and self.wallet.is_up_to_date():
self.wallet.save_verified_tx(write=True) # self.wallet.save_verified_tx(write=True)
@classmethod @classmethod
def hash_merkle_root(cls, merkle_branch: Sequence[str], tx_hash: str, leaf_pos_in_tree: int): def hash_merkle_root(cls, merkle_branch: Sequence[str], tx_hash: str, leaf_pos_in_tree: int):

67
electrum/wallet.py
View file

@ -276,7 +276,7 @@ class Abstract_Wallet(AddressSynchronizer):
return changed return changed
def set_fiat_value(self, txid, ccy, text, fx, value_sat): def set_fiat_value(self, txid, ccy, text, fx, value_sat):
if txid not in self.transactions: if txid not in self.db.list_transactions():
return return
# since fx is inserting the thousands separator, # since fx is inserting the thousands separator,
# and not util, also have fx remove it # and not util, also have fx remove it
@ -346,7 +346,8 @@ class Abstract_Wallet(AddressSynchronizer):
return [self.get_public_key(address)] return [self.get_public_key(address)]
def is_found(self): def is_found(self):
return self.history.values() != [[]] * len(self.history) return True
#return self.history.values() != [[]] * len(self.history)
def get_tx_info(self, tx): def get_tx_info(self, tx):
is_relevant, is_mine, v, fee = self.get_wallet_delta(tx) is_relevant, is_mine, v, fee = self.get_wallet_delta(tx)
@ -357,7 +358,7 @@ class Abstract_Wallet(AddressSynchronizer):
height = conf = timestamp = None height = conf = timestamp = None
tx_hash = tx.txid() tx_hash = tx.txid()
if tx.is_complete(): if tx.is_complete():
if tx_hash in self.transactions.keys(): if tx_hash in self.db.list_transactions():
label = self.get_label(tx_hash) label = self.get_label(tx_hash)
tx_mined_status = self.get_tx_height(tx_hash) tx_mined_status = self.get_tx_height(tx_hash)
height, conf = tx_mined_status.height, tx_mined_status.conf height, conf = tx_mined_status.height, tx_mined_status.conf
@ -369,7 +370,7 @@ class Abstract_Wallet(AddressSynchronizer):
elif height in (TX_HEIGHT_UNCONF_PARENT, TX_HEIGHT_UNCONFIRMED): elif height in (TX_HEIGHT_UNCONF_PARENT, TX_HEIGHT_UNCONFIRMED):
status = _('Unconfirmed') status = _('Unconfirmed')
if fee is None: if fee is None:
fee = self.tx_fees.get(tx_hash) fee = self.db.get_tx_fee(tx_hash)
if fee and self.network and self.network.config.has_fee_mempool(): if fee and self.network and self.network.config.has_fee_mempool():
size = tx.estimated_size() size = tx.estimated_size()
fee_per_byte = fee / size fee_per_byte = fee / size
@ -447,7 +448,7 @@ class Abstract_Wallet(AddressSynchronizer):
continue continue
if to_height is not None and height >= to_height: if to_height is not None and height >= to_height:
continue continue
tx = self.transactions.get(tx_hash) tx = self.db.get_transaction(tx_hash)
item = { item = {
'txid': tx_hash, 'txid': tx_hash,
'height': height, 'height': height,
@ -554,10 +555,9 @@ class Abstract_Wallet(AddressSynchronizer):
return label return label
def get_default_label(self, tx_hash): def get_default_label(self, tx_hash):
if self.txi.get(tx_hash) == {}: if not self.db.get_txi(tx_hash):
d = self.txo.get(tx_hash, {})
labels = [] labels = []
for addr in d.keys(): for addr in self.db.get_txo(tx_hash):
label = self.labels.get(addr) label = self.labels.get(addr)
if label: if label:
labels.append(label) labels.append(label)
@ -570,7 +570,7 @@ class Abstract_Wallet(AddressSynchronizer):
conf = tx_mined_info.conf conf = tx_mined_info.conf
timestamp = tx_mined_info.timestamp timestamp = tx_mined_info.timestamp
if conf == 0: if conf == 0:
tx = self.transactions.get(tx_hash) tx = self.db.get_transaction(tx_hash)
if not tx: if not tx:
return 2, 'unknown' return 2, 'unknown'
is_final = tx and tx.is_final() is_final = tx and tx.is_final()
@ -578,7 +578,7 @@ class Abstract_Wallet(AddressSynchronizer):
extra.append('rbf') extra.append('rbf')
fee = self.get_wallet_delta(tx)[3] fee = self.get_wallet_delta(tx)[3]
if fee is None: if fee is None:
fee = self.tx_fees.get(tx_hash) fee = self.db.get_tx_fee(tx_hash)
if fee is not None: if fee is not None:
size = tx.estimated_size() size = tx.estimated_size()
fee_per_byte = fee / size fee_per_byte = fee / size
@ -616,9 +616,11 @@ class Abstract_Wallet(AddressSynchronizer):
# tx should not be mined yet # tx should not be mined yet
if tx_mined_status.conf > 0: continue if tx_mined_status.conf > 0: continue
# tx should be "outgoing" from wallet # tx should be "outgoing" from wallet
if delta >= 0: continue if delta >= 0:
tx = self.transactions.get(tx_hash) continue
if not tx: continue tx = self.db.get_transaction(tx_hash)
if not tx:
continue
# is_mine outputs should not be spent yet # is_mine outputs should not be spent yet
# to avoid cancelling our own dependent transactions # to avoid cancelling our own dependent transactions
txid = tx.txid() txid = tx.txid()
@ -787,7 +789,7 @@ class Abstract_Wallet(AddressSynchronizer):
def address_is_old(self, address, age_limit=2): def address_is_old(self, address, age_limit=2):
age = -1 age = -1
h = self.history.get(address, []) h = self.db.get_addr_history(address)
for tx_hash, tx_height in h: for tx_hash, tx_height in h:
if tx_height <= 0: if tx_height <= 0:
tx_age = 0 tx_age = 0
@ -884,7 +886,7 @@ class Abstract_Wallet(AddressSynchronizer):
# First look up an input transaction in the wallet where it # First look up an input transaction in the wallet where it
# will likely be. If co-signing a transaction it may not have # will likely be. If co-signing a transaction it may not have
# all the input txs, in which case we ask the network. # all the input txs, in which case we ask the network.
tx = self.transactions.get(tx_hash, None) tx = self.db.get_transaction(tx_hash)
if not tx and self.network: if not tx and self.network:
try: try:
raw_tx = self.network.run_from_another_thread( raw_tx = self.network.run_from_another_thread(
@ -950,7 +952,7 @@ class Abstract_Wallet(AddressSynchronizer):
def get_unused_addresses(self): def get_unused_addresses(self):
# fixme: use slots from expired requests # fixme: use slots from expired requests
domain = self.get_receiving_addresses() domain = self.get_receiving_addresses()
return [addr for addr in domain if not self.history.get(addr) return [addr for addr in domain if not self.db.get_addr_history(addr)
and addr not in self.receive_requests.keys()] and addr not in self.receive_requests.keys()]
@check_returned_address @check_returned_address
@ -967,7 +969,7 @@ class Abstract_Wallet(AddressSynchronizer):
return return
choice = domain[0] choice = domain[0]
for addr in domain: for addr in domain:
if not self.history.get(addr): if not self.db.get_addr_history(addr):
if addr not in self.receive_requests.keys(): if addr not in self.receive_requests.keys():
return addr return addr
else: else:
@ -981,7 +983,7 @@ class Abstract_Wallet(AddressSynchronizer):
for txo, x in received.items(): for txo, x in received.items():
h, v, is_cb = x h, v, is_cb = x
txid, n = txo.split(':') txid, n = txo.split(':')
info = self.verified_tx.get(txid) info = self.db.get_verified_tx(txid)
if info: if info:
conf = local_height - info.height conf = local_height - info.height
else: else:
@ -1214,7 +1216,8 @@ class Abstract_Wallet(AddressSynchronizer):
def txin_value(self, txin): def txin_value(self, txin):
txid = txin['prevout_hash'] txid = txin['prevout_hash']
prev_n = txin['prevout_n'] prev_n = txin['prevout_n']
for address, d in self.txo.get(txid, {}).items(): for addr in self.db.get_txo(txid):
d = self.db.get_txo_addr(txid, addr)
for n, v, cb in d: for n, v, cb in d:
if n == prev_n: if n == prev_n:
return v return v
@ -1238,7 +1241,8 @@ class Abstract_Wallet(AddressSynchronizer):
""" Average acquisition price of the inputs of a transaction """ """ Average acquisition price of the inputs of a transaction """
input_value = 0 input_value = 0
total_price = 0 total_price = 0
for addr, d in self.txi.get(txid, {}).items(): for addr in self.db.get_txi(txid):
d = self.db.get_txi_addr(txid, addr)
for ser, v in d: for ser, v in d:
input_value += v input_value += v
total_price += self.coin_price(ser.split(':')[0], price_func, ccy, v) total_price += self.coin_price(ser.split(':')[0], price_func, ccy, v)
@ -1258,7 +1262,7 @@ class Abstract_Wallet(AddressSynchronizer):
result = self._coin_price_cache.get(cache_key, None) result = self._coin_price_cache.get(cache_key, None)
if result is not None: if result is not None:
return result return result
if self.txi.get(txid, {}) != {}: if self.db.get_txi(txid):
result = self.average_price(txid, price_func, ccy) * txin_value/Decimal(COIN) result = self.average_price(txid, price_func, ccy) * txin_value/Decimal(COIN)
self._coin_price_cache[cache_key] = result self._coin_price_cache[cache_key] = result
return result return result
@ -1398,7 +1402,8 @@ class Imported_Wallet(Simple_Wallet):
transactions_to_remove = set() # only referred to by this address transactions_to_remove = set() # only referred to by this address
transactions_new = set() # txs that are not only referred to by address transactions_new = set() # txs that are not only referred to by address
with self.lock: with self.lock:
for addr, details in self.history.items(): for addr in self.db.get_history():
details = self.db.get_addr_history(addr)
if addr == address: if addr == address:
for tx_hash, height in details: for tx_hash, height in details:
transactions_to_remove.add(tx_hash) transactions_to_remove.add(tx_hash)
@ -1406,14 +1411,14 @@ class Imported_Wallet(Simple_Wallet):
for tx_hash, height in details: for tx_hash, height in details:
transactions_new.add(tx_hash) transactions_new.add(tx_hash)
transactions_to_remove -= transactions_new transactions_to_remove -= transactions_new
self.history.pop(address, None) self.db.remove_history(address)
for tx_hash in transactions_to_remove: for tx_hash in transactions_to_remove:
self.remove_transaction(tx_hash) self.remove_transaction(tx_hash)
self.tx_fees.pop(tx_hash, None) self.db.remove_tx_fee(tx_hash)
self.verified_tx.pop(tx_hash, None) self.db.remove_verified_tx(tx_hash)
self.unverified_tx.pop(tx_hash, None) self.unverified_tx.pop(tx_hash, None)
self.transactions.pop(tx_hash, None) self.db.remove_transaction(tx_hash)
self.save_verified_tx() self.save_verified_tx()
self.save_transactions() self.save_transactions()
@ -1465,7 +1470,8 @@ class Imported_Wallet(Simple_Wallet):
self.add_address(addr) self.add_address(addr)
self.save_keystore() self.save_keystore()
self.save_addresses() self.save_addresses()
self.save_transactions(write=write_to_disk) if write_to_disk:
self.storage.write()
return good_addr, bad_keys return good_addr, bad_keys
def import_private_key(self, key: str, password: Optional[str]) -> str: def import_private_key(self, key: str, password: Optional[str]) -> str:
@ -1571,7 +1577,8 @@ class Deterministic_Wallet(Abstract_Wallet):
def num_unused_trailing_addresses(self, addresses): def num_unused_trailing_addresses(self, addresses):
k = 0 k = 0
for a in addresses[::-1]: for a in addresses[::-1]:
if self.history.get(a):break if a in self.db.get_history():
break
k = k + 1 k = k + 1
return k return k
@ -1582,7 +1589,7 @@ class Deterministic_Wallet(Abstract_Wallet):
addresses = self.get_receiving_addresses() addresses = self.get_receiving_addresses()
k = self.num_unused_trailing_addresses(addresses) k = self.num_unused_trailing_addresses(addresses)
for a in addresses[0:-k]: for a in addresses[0:-k]:
if self.history.get(a): if a in self.db.get_history():
n = 0 n = 0
else: else:
n += 1 n += 1
@ -1641,7 +1648,7 @@ class Deterministic_Wallet(Abstract_Wallet):
return False return False
prev_addresses = addr_list[max(0, i - limit):max(0, i)] prev_addresses = addr_list[max(0, i - limit):max(0, i)]
for addr in prev_addresses: for addr in prev_addresses:
if self.history.get(addr): if addr in self.db.get_history():
return False return False
return True return True