4946546086641664

Committed 24 Sep 2025 01:50PM UTC coverage: 61.311% (+0.01%) from 61.3%

Build # 4946546086641664

Build Type

push

CirrusCI

Committed by

web-flow

Commit Message

Merge pull request #10216 from SomberNight/202509_adb_spv

adb: change API of util.TxMinedInfo: height() is now always SPV-ed

Coverage Stats

45 of 71 new or added lines in 12 files covered. (63.38%)

2 existing lines in 1 file now uncovered.

22839 of 37251 relevant lines covered (61.31%)

2.45 hits per line

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

38.07

/electrum/lnwatcher.py

# Copyright (C) 2018 The Electrum developers
# Distributed under the MIT software license, see the accompanying
# file LICENCE or http://www.opensource.org/licenses/mit-license.php

from typing import TYPE_CHECKING, Optional, Dict, Callable, Awaitable

from . import util
from .util import TxMinedInfo, BelowDustLimit, NoDynamicFeeEstimates
from .util import EventListener, event_listener, log_exceptions, ignore_exceptions
from .transaction import Transaction, TxOutpoint
from .logging import Logger
from .address_synchronizer import TX_HEIGHT_LOCAL
from .lnutil import REDEEM_AFTER_DOUBLE_SPENT_DELAY


if TYPE_CHECKING:
    from .network import Network
    from .lnsweep import SweepInfo
    from .lnworker import LNWallet
    from .lnchannel import AbstractChannel


class LNWatcher(Logger, EventListener):

    def __init__(self, lnworker: 'LNWallet'):
        self.lnworker = lnworker
        Logger.__init__(self)
        self.adb = lnworker.wallet.adb
        self.config = lnworker.config
        self.callbacks = {}  # type: Dict[str, Callable[[], Awaitable[None]]]  # address -> lambda function
        self.network = None
        self.register_callbacks()
        self._pending_force_closes = set()

    def start_network(self, network: 'Network'):
        self.network = network

    def stop(self):
        self.unregister_callbacks()

    def remove_callback(self, address: str) -> None:
        self.callbacks.pop(address, None)

    def add_callback(
        self,
        address: str,
        callback: Callable[[], Awaitable[None]],
        *,
        subscribe: bool = True,
    ) -> None:
        if subscribe:
            self.adb.add_address(address)
        self.callbacks[address] = callback

    async def trigger_callbacks(self, *, requires_synchronizer: bool = True):
        if requires_synchronizer and not self.adb.synchronizer:
            self.logger.info("synchronizer not set yet")
            return
        for address, callback in list(self.callbacks.items()):
            try:
                await callback()
            except Exception:
                self.logger.exception(f"LNWatcher callback failed {address=}")
        # send callback to GUI
        util.trigger_callback('wallet_updated', self.lnworker.wallet)

    @event_listener
    async def on_event_blockchain_updated(self, *args):
        await self.trigger_callbacks()

    @event_listener
    async def on_event_adb_added_tx(self, adb, tx_hash, tx):
        # called if we add local tx
        if adb != self.adb:
            return
        await self.trigger_callbacks()

    @event_listener
    async def on_event_adb_added_verified_tx(self, adb, tx_hash):
        if adb != self.adb:
            return
        await self.trigger_callbacks()

    @event_listener
    async def on_event_adb_set_up_to_date(self, adb):
        if adb != self.adb:
            return
        await self.trigger_callbacks()

    def add_channel(self, chan: 'AbstractChannel') -> None:
        outpoint = chan.funding_outpoint.to_str()
        address = chan.get_funding_address()
        callback = lambda: self.check_onchain_situation(address, outpoint)
        self.add_callback(address, callback, subscribe=chan.need_to_subscribe())

    @ignore_exceptions
    @log_exceptions
    async def check_onchain_situation(self, address: str, funding_outpoint: str) -> None:
        # early return if address has not been added yet
        if not self.adb.is_mine(address):
            return
        # inspect_tx_candidate might have added new addresses, in which case we return early
        # note: maybe we should wait until adb.is_up_to_date... (?)
        funding_txid = funding_outpoint.split(':')[0]
        funding_height = self.adb.get_tx_height(funding_txid)
        closing_txid = self.adb.get_spender(funding_outpoint)
        closing_height = self.adb.get_tx_height(closing_txid)
        if closing_txid:
            closing_tx = self.adb.get_transaction(closing_txid)
            if closing_tx:
                keep_watching = await self.sweep_commitment_transaction(funding_outpoint, closing_tx)
            else:
                self.logger.info(f"channel {funding_outpoint} closed by {closing_txid}. still waiting for tx itself...")
                keep_watching = True
        else:
            keep_watching = True
        await self.update_channel_state(
            funding_outpoint=funding_outpoint,
            funding_txid=funding_txid,
            funding_height=funding_height,
            closing_txid=closing_txid,
            closing_height=closing_height,
            keep_watching=keep_watching)

    def diagnostic_name(self):
        return f"{self.lnworker.wallet.diagnostic_name()}-LNW"

    async def update_channel_state(
            self, *, funding_outpoint: str, funding_txid: str,
            funding_height: TxMinedInfo, closing_txid: str,
            closing_height: TxMinedInfo, keep_watching: bool) -> None:
        chan = self.lnworker.channel_by_txo(funding_outpoint)
        if not chan:
            return
        chan.update_onchain_state(
            funding_txid=funding_txid,
            funding_height=funding_height,
            closing_txid=closing_txid,
            closing_height=closing_height,
            keep_watching=keep_watching)
        if closing_height.conf > 0:
            self._pending_force_closes.discard(chan)
        await self.lnworker.handle_onchain_state(chan)

    async def sweep_commitment_transaction(self, funding_outpoint: str, closing_tx: Transaction) -> bool:
        """This function is called when a channel was closed. In this case
        we need to check for redeemable outputs of the commitment transaction
        or spenders down the line (HTLC-timeout/success transactions).

        Returns whether we should continue to monitor.

        Side-effects:
          - sets defaults labels
          - populates wallet._accounting_addresses
        """
        assert closing_tx
        chan = self.lnworker.channel_by_txo(funding_outpoint)
        if not chan:
            return False
        if not chan.need_to_subscribe():
            return False
        self.logger.info(f'sweep_commitment_transaction {funding_outpoint}')
        # detect who closed and get information about how to claim outputs
        is_local_ctx, sweep_info_dict = chan.get_ctx_sweep_info(closing_tx)
        # note: we need to keep watching *at least* until the closing tx is deeply mined,
        #       possibly longer if there are TXOs to sweep
        keep_watching = not self.adb.is_deeply_mined(closing_tx.txid())
        # create and broadcast transactions
        for prevout, sweep_info in sweep_info_dict.items():
            prev_txid, prev_index = prevout.split(':')
            name = sweep_info.name + ' ' + chan.get_id_for_log()
            self.lnworker.wallet.set_default_label(prevout, name)
            if not self.adb.get_transaction(prev_txid):
                # do not keep watching if prevout does not exist
                self.logger.info(f'prevout does not exist for {name}: {prevout}')
                continue
            watch_sweep_info = self.maybe_redeem(sweep_info)
            spender_txid = self.adb.get_spender(prevout)  # note: LOCAL spenders don't count
            spender_tx = self.adb.get_transaction(spender_txid) if spender_txid else None
            if spender_tx:
                # the spender might be the remote, revoked or not
                htlc_sweepinfo = chan.maybe_sweep_htlcs(closing_tx, spender_tx)
                for prevout2, htlc_sweep_info in htlc_sweepinfo.items():
                    watch_htlc_sweep_info = self.maybe_redeem(htlc_sweep_info)
                    htlc_tx_spender = self.adb.get_spender(prevout2)
                    self.lnworker.wallet.set_default_label(prevout2, htlc_sweep_info.name)
                    if htlc_tx_spender:
                        keep_watching |= not self.adb.is_deeply_mined(htlc_tx_spender)
                        self.maybe_add_accounting_address(htlc_tx_spender, htlc_sweep_info)
                    else:
                        keep_watching |= watch_htlc_sweep_info
                keep_watching |= not self.adb.is_deeply_mined(spender_txid)
                self.maybe_extract_preimage(chan, spender_tx, prevout)
                self.maybe_add_accounting_address(spender_txid, sweep_info)
            else:
                keep_watching |= watch_sweep_info
            self.maybe_add_pending_forceclose(
                chan=chan,
                spender_txid=spender_txid,
                is_local_ctx=is_local_ctx,
                sweep_info=sweep_info,
            )
        return keep_watching

    def get_pending_force_closes(self):
        return self._pending_force_closes

    def maybe_redeem(self, sweep_info: 'SweepInfo') -> bool:
        """ returns 'keep_watching' """
        try:
            self.lnworker.wallet.txbatcher.add_sweep_input('lnwatcher', sweep_info)
        except BelowDustLimit:
            # utxo is considered dust at *current* fee estimates.
            # but maybe the fees atm are very high? We will retry later.
            pass
        except NoDynamicFeeEstimates:
            pass  # will retry later
        if sweep_info.is_anchor():
            return False
        return True

    def maybe_extract_preimage(self, chan: 'AbstractChannel', spender_tx: Transaction, prevout: str):
        if not spender_tx.is_complete():
            self.logger.info('spender tx is unsigned')
            return
        txin_idx = spender_tx.get_input_idx_that_spent_prevout(TxOutpoint.from_str(prevout))
        assert txin_idx is not None
        spender_txin = spender_tx.inputs()[txin_idx]
        chan.extract_preimage_from_htlc_txin(
            spender_txin,
            is_deeply_mined=self.adb.is_deeply_mined(spender_tx.txid()),
        )

    def maybe_add_accounting_address(self, spender_txid: str, sweep_info: 'SweepInfo'):
        spender_tx = self.adb.get_transaction(spender_txid) if spender_txid else None
        if not spender_tx:
            return
        for i, txin in enumerate(spender_tx.inputs()):
            if txin.prevout == sweep_info.txin.prevout:
                break
        else:
            return
        if sweep_info.name in ['offered-htlc', 'received-htlc']:
            # always consider ours
            pass
        else:
            witness = txin.witness_elements()
            for sig in witness:
                # fixme: verify sig is ours
                witness2 = sweep_info.txin.make_witness(sig)
                if txin.witness == witness2:
                    break
            else:
                self.logger.info(f"signature not found {sweep_info.name}, {txin.prevout.to_str()}")
                return
        self.logger.info(f'adding txin address {sweep_info.name}, {txin.prevout.to_str()}')
        prev_txid, prev_index = txin.prevout.to_str().split(':')
        prev_tx = self.adb.get_transaction(prev_txid)
        txout = prev_tx.outputs()[int(prev_index)]
        self.lnworker.wallet._accounting_addresses.add(txout.address)

    def maybe_add_pending_forceclose(
        self,
        *,
        chan: 'AbstractChannel',
        spender_txid: Optional[str],
        is_local_ctx: bool,
        sweep_info: 'SweepInfo',
    ) -> None:
        """Adds chan into set of ongoing force-closures if the user should keep the wallet open, waiting for it.
        (we are waiting for ctx to be confirmed and there are received htlcs)
        """
        if is_local_ctx and sweep_info.name == 'received-htlc':
            cltv = sweep_info.cltv_abs
            assert cltv is not None, f"missing cltv for {sweep_info}"
            if self.adb.get_local_height() > cltv + REDEEM_AFTER_DOUBLE_SPENT_DELAY:
                # We had plenty of time to sweep. The remote also had time to time out the htlc.
                # Maybe its value has been ~dust at current and past fee levels (every time we checked).
                # We should not keep warning the user forever.
                return
            tx_mined_status = self.adb.get_tx_height(spender_txid)
            if tx_mined_status.height() == TX_HEIGHT_LOCAL:
                self._pending_force_closes.add(chan)

1	# Copyright (C) 2018 The Electrum developers
2	# Distributed under the MIT software license, see the accompanying
3	# file LICENCE or http://www.opensource.org/licenses/mit-license.php
4
5	from typing import TYPE_CHECKING, Optional, Dict, Callable, Awaitable	4✔
6
7	from . import util	4✔
8	from .util import TxMinedInfo, BelowDustLimit, NoDynamicFeeEstimates	4✔
9	from .util import EventListener, event_listener, log_exceptions, ignore_exceptions	4✔
10	from .transaction import Transaction, TxOutpoint	4✔
11	from .logging import Logger	4✔
12	from .address_synchronizer import TX_HEIGHT_LOCAL	4✔
13	from .lnutil import REDEEM_AFTER_DOUBLE_SPENT_DELAY	4✔
14
15
16	if TYPE_CHECKING:	2✔
17	from .network import Network
18	from .lnsweep import SweepInfo
19	from .lnworker import LNWallet
20	from .lnchannel import AbstractChannel
21
22
23	class LNWatcher(Logger, EventListener):	4✔
24
25	def __init__(self, lnworker: 'LNWallet'):	4✔
26	self.lnworker = lnworker	4✔
27	Logger.__init__(self)	4✔
28	self.adb = lnworker.wallet.adb	4✔
29	self.config = lnworker.config	4✔
30	self.callbacks = {} # type: Dict[str, Callable[[], Awaitable[None]]] # address -> lambda function	4✔
31	self.network = None	4✔
32	self.register_callbacks()	4✔
33	self._pending_force_closes = set()	4✔
34
35	def start_network(self, network: 'Network'):	4✔
36	self.network = network	×
37
38	def stop(self):	4✔
39	self.unregister_callbacks()	×
40
41	def remove_callback(self, address: str) -> None:	4✔
42	self.callbacks.pop(address, None)	×
43
44	def add_callback(	4✔
45	self,
46	address: str,
47	callback: Callable[[], Awaitable[None]],
48	*,
49	subscribe: bool = True,
50	) -> None:
51	if subscribe:	4✔
52	self.adb.add_address(address)	4✔
53	self.callbacks[address] = callback	4✔
54
55	async def trigger_callbacks(self, *, requires_synchronizer: bool = True):	4✔
56	if requires_synchronizer and not self.adb.synchronizer:	4✔
57	self.logger.info("synchronizer not set yet")	4✔
58	return	4✔
59	for address, callback in list(self.callbacks.items()):	4✔
60	try:	×
61	await callback()	×
62	except Exception:	×
63	self.logger.exception(f"LNWatcher callback failed {address=}")	×
64	# send callback to GUI
65	util.trigger_callback('wallet_updated', self.lnworker.wallet)	4✔
66
67	@event_listener	4✔
68	async def on_event_blockchain_updated(self, *args):	4✔
69	await self.trigger_callbacks()	4✔
70
71	@event_listener	4✔
72	async def on_event_adb_added_tx(self, adb, tx_hash, tx):	4✔
73	# called if we add local tx
74	if adb != self.adb:	4✔
75	return	4✔
76	await self.trigger_callbacks()	4✔
77
78	@event_listener	4✔
79	async def on_event_adb_added_verified_tx(self, adb, tx_hash):	4✔
80	if adb != self.adb:	4✔
81	return	4✔
82	await self.trigger_callbacks()	4✔
83
84	@event_listener	4✔
85	async def on_event_adb_set_up_to_date(self, adb):	4✔
86	if adb != self.adb:	4✔
87	return	4✔
88	await self.trigger_callbacks()	4✔
89
90	def add_channel(self, chan: 'AbstractChannel') -> None:	4✔
91	outpoint = chan.funding_outpoint.to_str()	4✔
92	address = chan.get_funding_address()	4✔
93	callback = lambda: self.check_onchain_situation(address, outpoint)	4✔
94	self.add_callback(address, callback, subscribe=chan.need_to_subscribe())	4✔
95
96	@ignore_exceptions	4✔
97	@log_exceptions	4✔
98	async def check_onchain_situation(self, address: str, funding_outpoint: str) -> None:	4✔
99	# early return if address has not been added yet
100	if not self.adb.is_mine(address):	×
101	return	×
102	# inspect_tx_candidate might have added new addresses, in which case we return early
103	# note: maybe we should wait until adb.is_up_to_date... (?)
104	funding_txid = funding_outpoint.split(':')[0]	×
105	funding_height = self.adb.get_tx_height(funding_txid)	×
106	closing_txid = self.adb.get_spender(funding_outpoint)	×
107	closing_height = self.adb.get_tx_height(closing_txid)	×
108	if closing_txid:	×
109	closing_tx = self.adb.get_transaction(closing_txid)	×
110	if closing_tx:	×
111	keep_watching = await self.sweep_commitment_transaction(funding_outpoint, closing_tx)	×
112	else:
113	self.logger.info(f"channel {funding_outpoint} closed by {closing_txid}. still waiting for tx itself...")	×
114	keep_watching = True	×
115	else:
116	keep_watching = True	×
117	await self.update_channel_state(	×
118	funding_outpoint=funding_outpoint,
119	funding_txid=funding_txid,
120	funding_height=funding_height,
121	closing_txid=closing_txid,
122	closing_height=closing_height,
123	keep_watching=keep_watching)
124
125	def diagnostic_name(self):	4✔
126	return f"{self.lnworker.wallet.diagnostic_name()}-LNW"	4✔
127
128	async def update_channel_state(	4✔
129	self, *, funding_outpoint: str, funding_txid: str,
130	funding_height: TxMinedInfo, closing_txid: str,
131	closing_height: TxMinedInfo, keep_watching: bool) -> None:
132	chan = self.lnworker.channel_by_txo(funding_outpoint)	×
133	if not chan:	×
134	return	×
135	chan.update_onchain_state(	×
136	funding_txid=funding_txid,
137	funding_height=funding_height,
138	closing_txid=closing_txid,
139	closing_height=closing_height,
140	keep_watching=keep_watching)
141	if closing_height.conf > 0:	×
142	self._pending_force_closes.discard(chan)	×
143	await self.lnworker.handle_onchain_state(chan)	×
144
145	async def sweep_commitment_transaction(self, funding_outpoint: str, closing_tx: Transaction) -> bool:	4✔
146	"""This function is called when a channel was closed. In this case
147	we need to check for redeemable outputs of the commitment transaction
148	or spenders down the line (HTLC-timeout/success transactions).
149
150	Returns whether we should continue to monitor.
151
152	Side-effects:
153	- sets defaults labels
154	- populates wallet._accounting_addresses
155	"""
156	assert closing_tx	×
157	chan = self.lnworker.channel_by_txo(funding_outpoint)	×
158	if not chan:	×
159	return False	×
160	if not chan.need_to_subscribe():	×
161	return False	×
162	self.logger.info(f'sweep_commitment_transaction {funding_outpoint}')	×
163	# detect who closed and get information about how to claim outputs
164	is_local_ctx, sweep_info_dict = chan.get_ctx_sweep_info(closing_tx)	×
165	# note: we need to keep watching at least until the closing tx is deeply mined,
166	# possibly longer if there are TXOs to sweep
167	keep_watching = not self.adb.is_deeply_mined(closing_tx.txid())	×
168	# create and broadcast transactions
169	for prevout, sweep_info in sweep_info_dict.items():	×
170	prev_txid, prev_index = prevout.split(':')	×
171	name = sweep_info.name + ' ' + chan.get_id_for_log()	×
172	self.lnworker.wallet.set_default_label(prevout, name)	×
173	if not self.adb.get_transaction(prev_txid):	×
174	# do not keep watching if prevout does not exist
175	self.logger.info(f'prevout does not exist for {name}: {prevout}')	×
176	continue	×
177	watch_sweep_info = self.maybe_redeem(sweep_info)	×
178	spender_txid = self.adb.get_spender(prevout) # note: LOCAL spenders don't count	×
179	spender_tx = self.adb.get_transaction(spender_txid) if spender_txid else None	×
180	if spender_tx:	×
181	# the spender might be the remote, revoked or not
182	htlc_sweepinfo = chan.maybe_sweep_htlcs(closing_tx, spender_tx)	×
183	for prevout2, htlc_sweep_info in htlc_sweepinfo.items():	×
184	watch_htlc_sweep_info = self.maybe_redeem(htlc_sweep_info)	×
185	htlc_tx_spender = self.adb.get_spender(prevout2)	×
186	self.lnworker.wallet.set_default_label(prevout2, htlc_sweep_info.name)	×
187	if htlc_tx_spender:	×
188	keep_watching \|= not self.adb.is_deeply_mined(htlc_tx_spender)	×
189	self.maybe_add_accounting_address(htlc_tx_spender, htlc_sweep_info)	×
190	else:
191	keep_watching \|= watch_htlc_sweep_info	×
192	keep_watching \|= not self.adb.is_deeply_mined(spender_txid)	×
193	self.maybe_extract_preimage(chan, spender_tx, prevout)	×
194	self.maybe_add_accounting_address(spender_txid, sweep_info)	×
195	else:
196	keep_watching \|= watch_sweep_info	×
197	self.maybe_add_pending_forceclose(	×
198	chan=chan,
199	spender_txid=spender_txid,
200	is_local_ctx=is_local_ctx,
201	sweep_info=sweep_info,
202	)
203	return keep_watching	×
204
205	def get_pending_force_closes(self):	4✔
206	return self._pending_force_closes	×
207
208	def maybe_redeem(self, sweep_info: 'SweepInfo') -> bool:	4✔
209	""" returns 'keep_watching' """
210	try:	×
211	self.lnworker.wallet.txbatcher.add_sweep_input('lnwatcher', sweep_info)	×
212	except BelowDustLimit:	×
213	# utxo is considered dust at current fee estimates.
214	# but maybe the fees atm are very high? We will retry later.
215	pass	×
216	except NoDynamicFeeEstimates:	×
217	pass # will retry later	×
218	if sweep_info.is_anchor():	×
219	return False	×
220	return True	×
221
222	def maybe_extract_preimage(self, chan: 'AbstractChannel', spender_tx: Transaction, prevout: str):	4✔
223	if not spender_tx.is_complete():	×
224	self.logger.info('spender tx is unsigned')	×
225	return	×
226	txin_idx = spender_tx.get_input_idx_that_spent_prevout(TxOutpoint.from_str(prevout))	×
227	assert txin_idx is not None	×
228	spender_txin = spender_tx.inputs()[txin_idx]	×
229	chan.extract_preimage_from_htlc_txin(	×
230	spender_txin,
231	is_deeply_mined=self.adb.is_deeply_mined(spender_tx.txid()),
232	)
233
234	def maybe_add_accounting_address(self, spender_txid: str, sweep_info: 'SweepInfo'):	4✔
235	spender_tx = self.adb.get_transaction(spender_txid) if spender_txid else None	×
236	if not spender_tx:	×
237	return	×
238	for i, txin in enumerate(spender_tx.inputs()):	×
239	if txin.prevout == sweep_info.txin.prevout:	×
240	break	×
241	else:
242	return	×
243	if sweep_info.name in ['offered-htlc', 'received-htlc']:	×
244	# always consider ours
245	pass	×
246	else:
247	witness = txin.witness_elements()	×
248	for sig in witness:	×
249	# fixme: verify sig is ours
250	witness2 = sweep_info.txin.make_witness(sig)	×
251	if txin.witness == witness2:	×
252	break	×
253	else:
254	self.logger.info(f"signature not found {sweep_info.name}, {txin.prevout.to_str()}")	×
255	return	×
256	self.logger.info(f'adding txin address {sweep_info.name}, {txin.prevout.to_str()}')	×
257	prev_txid, prev_index = txin.prevout.to_str().split(':')	×
258	prev_tx = self.adb.get_transaction(prev_txid)	×
259	txout = prev_tx.outputs()[int(prev_index)]	×
260	self.lnworker.wallet._accounting_addresses.add(txout.address)	×
261
262	def maybe_add_pending_forceclose(	4✔
263	self,
264	*,
265	chan: 'AbstractChannel',
266	spender_txid: Optional[str],
267	is_local_ctx: bool,
268	sweep_info: 'SweepInfo',
269	) -> None:
270	"""Adds chan into set of ongoing force-closures if the user should keep the wallet open, waiting for it.
271	(we are waiting for ctx to be confirmed and there are received htlcs)
272	"""
273	if is_local_ctx and sweep_info.name == 'received-htlc':	×
274	cltv = sweep_info.cltv_abs	×
275	assert cltv is not None, f"missing cltv for {sweep_info}"	×
276	if self.adb.get_local_height() > cltv + REDEEM_AFTER_DOUBLE_SPENT_DELAY:	×
277	# We had plenty of time to sweep. The remote also had time to time out the htlc.
278	# Maybe its value has been ~dust at current and past fee levels (every time we checked).
279	# We should not keep warning the user forever.
280	return	×
281	tx_mined_status = self.adb.get_tx_height(spender_txid)	×
NEW 282	if tx_mined_status.height() == TX_HEIGHT_LOCAL:	×
283	self._pending_force_closes.add(chan)	×

spesmilo / electrum / 4946546086641664

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

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