Merge pull request #9298 from spesmilo/batch_payments_manager

wallet: RBF batch payments manager

Author: ecdsa

electrum/gui/qt/main_window.py

@@ -461,6 +461,11 @@ def on_event_channel(self, *args):
     def on_event_banner(self, *args):
         self.console.showMessage(args[0])
 
+    @qt_event_listener
+    def on_event_adb_set_future_tx(self, adb, txid):
+        if adb == self.wallet.adb:
+            self.history_model.refresh('set_future_tx')
+
     @qt_event_listener
     def on_event_verified(self, *args):
         wallet, tx_hash, tx_mined_status = args

electrum/lnsweep.py

@@ -43,6 +43,8 @@ class SweepInfo(NamedTuple):
     cltv_abs: Optional[int] # set to None only if the script has no cltv
     txin: PartialTxInput
     txout: Optional[PartialTxOutput]  # only for first-stage htlc tx
+    can_be_batched: bool # todo: this could be more fine-grained
+
 
 def sweep_their_ctx_watchtower(
         chan: 'Channel',
@@ -251,7 +253,8 @@ def justice_txin(output_idx):
                 csv_delay=0,
                 cltv_abs=None,
                 txin=txin,
-                txout=None
+                txout=None,
+                can_be_batched=False,
             )
     return index_to_sweepinfo
 
@@ -329,6 +332,7 @@ def sweep_our_ctx(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
 
     # to_local
@@ -350,6 +354,7 @@ def sweep_our_ctx(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
     we_breached = ctn < chan.get_oldest_unrevoked_ctn(LOCAL)
     if we_breached:
@@ -384,7 +389,11 @@ def txs_htlc(
                 csv_delay=0,
                 cltv_abs=htlc_tx.locktime,
                 txin=htlc_tx.inputs()[0],
-                txout=htlc_tx.outputs()[0])
+                txout=htlc_tx.outputs()[0],
+                can_be_batched=False, # both parties can spend
+                # actually, we might want to batch depending on the context
+                # f(amount in htlc, remaining_time, number of available utxos for anchors)
+            )
         else:
             # second-stage
             address = bitcoin.script_to_p2wsh(htlctx_witness_script)
@@ -404,6 +413,9 @@ def txs_htlc(
                         cltv_abs=0,
                         txin=sweep_txin,
                         txout=None,
+                        # this is safe to batch, we are the only ones who can spend
+                        # (assuming we did not broadcast a revoked state)
+                        can_be_batched=True,
                     )
 
     # offered HTLCs, in our ctx --> "timeout"
@@ -541,6 +553,7 @@ def sweep_their_ctx_to_remote_backup(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
 
     # to_remote
@@ -562,6 +575,7 @@ def sweep_their_ctx_to_remote_backup(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
     return txs
 
@@ -619,6 +633,7 @@ def sweep_their_ctx(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
 
     # to_local is handled by lnwatcher
@@ -631,6 +646,7 @@ def sweep_their_ctx(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=False,
             )
 
     # to_remote
@@ -656,12 +672,14 @@ def sweep_their_ctx(
                     our_payment_privkey=our_payment_privkey,
                     has_anchors=chan.has_anchors()
             ):
+                # todo: we might not want to sweep this at all, if we add it to the wallet addresses
                 txs[prevout] = SweepInfo(
                     name='their_ctx_to_remote',
                     csv_delay=csv_delay,
                     cltv_abs=None,
                     txin=txin,
                     txout=None,
+                    can_be_batched=True,
                 )
 
     # HTLCs
@@ -701,6 +719,7 @@ def tx_htlc(
                 cltv_abs=cltv_abs,
                 txin=txin,
                 txout=None,
+                can_be_batched=False,
             )
     # received HTLCs, in their ctx --> "timeout"
     # offered HTLCs, in their ctx --> "success"

electrum/lnwatcher.py

@@ -2,20 +2,14 @@
 # Distributed under the MIT software license, see the accompanying
 # file LICENCE or http://www.opensource.org/licenses/mit-license.php
 
-from typing import NamedTuple, Iterable, TYPE_CHECKING
-import copy
-import asyncio
+from typing import TYPE_CHECKING
 from enum import IntEnum, auto
-from typing import NamedTuple, Dict
 
-from . import util
-from .util import log_exceptions, ignore_exceptions, TxMinedInfo
+from .util import log_exceptions, ignore_exceptions, TxMinedInfo, BelowDustLimit
 from .util import EventListener, event_listener
 from .address_synchronizer import AddressSynchronizer, TX_HEIGHT_LOCAL, TX_HEIGHT_UNCONF_PARENT, TX_HEIGHT_UNCONFIRMED, TX_HEIGHT_FUTURE
-from .transaction import Transaction, TxOutpoint, PartialTransaction
+from .transaction import Transaction, TxOutpoint
 from .logging import Logger
-from .bitcoin import dust_threshold
-from .fee_policy import FeePolicy
 
 
 if TYPE_CHECKING:
@@ -46,7 +40,6 @@ def __init__(self, adb: 'AddressSynchronizer', network: 'Network'):
         self.register_callbacks()
         # status gets populated when we run
         self.channel_status = {}
-        self.fee_policy = FeePolicy('eta:2')
 
     async def stop(self):
         self.unregister_callbacks()
@@ -75,6 +68,13 @@ def add_callback(self, address, callback):
     async def on_event_blockchain_updated(self, *args):
         await self.trigger_callbacks()
 
+    @event_listener
+    async def on_event_wallet_updated(self, wallet):
+        # called if we add local tx
+        if wallet.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:
@@ -141,6 +141,10 @@ def get_spender(self, outpoint) -> str:
         """
         prev_txid, index = outpoint.split(':')
         spender_txid = self.adb.db.get_spent_outpoint(prev_txid, int(index))
+        # discard local spenders
+        tx_mined_status = self.adb.get_tx_height(spender_txid)
+        if tx_mined_status.height in [TX_HEIGHT_LOCAL, TX_HEIGHT_FUTURE]:
+            spender_txid = None
         if not spender_txid:
             return
         spender_tx = self.adb.get_transaction(spender_txid)
@@ -211,18 +215,6 @@ async def update_channel_state(self, *, funding_outpoint: str, funding_txid: str
             keep_watching=keep_watching)
         await self.lnworker.handle_onchain_state(chan)
 
-    def is_dust(self, sweep_info):
-        if sweep_info.name in ['local_anchor', 'remote_anchor']:
-            return False
-        if sweep_info.txout is not None:
-            return False
-        value = sweep_info.txin._trusted_value_sats
-        witness_size = len(sweep_info.txin.make_witness(71*b'\x00'))
-        tx_size_vbytes = 84 + witness_size//4     # assumes no batching, sweep to p2wpkh
-        self.logger.info(f'{sweep_info.name} size = {tx_size_vbytes}')
-        fee = self.fee_policy.estimate_fee(tx_size_vbytes, network=self.network, allow_fallback_to_static_rates=True)
-        return value - fee <= dust_threshold()
-
     @log_exceptions
     async def sweep_commitment_transaction(self, funding_outpoint, closing_tx) -> bool:
         """This function is called when a channel was closed. In this case
@@ -235,19 +227,16 @@ async def sweep_commitment_transaction(self, funding_outpoint, closing_tx) -> bo
             return False
         # detect who closed and get information about how to claim outputs
         sweep_info_dict = chan.sweep_ctx(closing_tx)
-        self.logger.info(f"do_breach_remedy: {[x.name for x in sweep_info_dict.values()]}")
+        #self.logger.info(f"do_breach_remedy: {[x.name for x in sweep_info_dict.values()]}")
         keep_watching = False if sweep_info_dict else not self.is_deeply_mined(closing_tx.txid())
-
         # create and broadcast transactions
         for prevout, sweep_info in sweep_info_dict.items():
-            if self.is_dust(sweep_info):
-                continue
             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}: {prev_txid}')
+                self.logger.info(f'prevout does not exist for {name}: {prevout}')
                 continue
             spender_txid = self.get_spender(prevout)
             spender_tx = self.adb.get_transaction(spender_txid) if spender_txid else None
@@ -260,115 +249,20 @@ async def sweep_commitment_transaction(self, funding_outpoint, closing_tx) -> bo
                     if htlc_tx_spender:
                         keep_watching |= not self.is_deeply_mined(htlc_tx_spender)
                     else:
-                        keep_watching = True
-                    await self.maybe_redeem(prevout2, htlc_sweep_info, name)
+                        keep_watching |= self.maybe_redeem(htlc_sweep_info)
                 keep_watching |= not self.is_deeply_mined(spender_txid)
                 self.maybe_extract_preimage(chan, spender_tx, prevout)
             else:
-                keep_watching = True
-            # broadcast or maybe update our own tx
-            await self.maybe_redeem(prevout, sweep_info, name)
-
+                keep_watching |= self.maybe_redeem(sweep_info)
         return keep_watching
 
-    def get_redeem_tx(self, prevout: str, sweep_info: 'SweepInfo', name: str):
-        # check if redeem tx needs to be updated
-        # if it is in the mempool, we need to check fee rise
-        txid = self.get_spender(prevout)
-        old_tx = self.adb.get_transaction(txid)
-        assert old_tx is not None or txid is None
-        tx_depth = self.get_tx_mined_depth(txid) if txid else None
-        if txid and tx_depth not in [TxMinedDepth.FREE, TxMinedDepth.MEMPOOL]:
-            assert old_tx is not None
-            return old_tx, None
-        # fixme: deepcopy is needed because tx.serialize() is destructive
-        inputs = [copy.deepcopy(sweep_info.txin)]
-        outputs = [sweep_info.txout] if sweep_info.txout else []
-        if sweep_info.name == 'first-stage-htlc':
-            new_tx = PartialTransaction.from_io(inputs, outputs, locktime=sweep_info.cltv_abs, version=2)
-            self.lnworker.wallet.sign_transaction(new_tx, password=None, ignore_warnings=True)
-        else:
-            # password is needed for 1st stage htlc tx with anchors because we add inputs
-            password = self.lnworker.wallet.get_unlocked_password()
-            new_tx = self.lnworker.wallet.create_transaction(
-                fee_policy = self.fee_policy,
-                inputs = inputs,
-                outputs = outputs,
-                password = password,
-                locktime = sweep_info.cltv_abs,
-                BIP69_sort=False,
-            )
-        if new_tx is None:
-            self.logger.info(f'{name} could not claim output: {prevout}, dust')
-            assert old_tx is not None
-            return old_tx, None
-        if txid is None:
-            return None, new_tx
-        elif tx_depth == TxMinedDepth.MEMPOOL:
-            delta = new_tx.get_fee() - self.adb.get_tx_fee(txid)
-            if delta > 1:
-                self.logger.info(f'increasing fee of mempool tx {name}: {prevout}')
-                return old_tx, new_tx
-            else:
-                assert old_tx is not None
-                return old_tx, None
-        elif tx_depth == TxMinedDepth.FREE:
-            # return new tx, even if it is equal to old_tx,
-            # because we need to test if it can be broadcast
-            return old_tx, new_tx
-        else:
-            assert old_tx is not None
-            return old_tx, None
-
-    async def maybe_redeem(self, prevout, sweep_info: 'SweepInfo', name: str) -> None:
-        old_tx, new_tx = self.get_redeem_tx(prevout, sweep_info, name)
-        if new_tx is None:
-            return
-        prev_txid, prev_index = prevout.split(':')
-        can_broadcast = True
-        local_height = self.network.get_local_height()
-        if sweep_info.cltv_abs:
-            wanted_height = sweep_info.cltv_abs
-            if wanted_height - local_height > 0:
-                can_broadcast = False
-                # self.logger.debug(f"pending redeem for {prevout}. waiting for {name}: CLTV ({local_height=}, {wanted_height=})")
-        if sweep_info.csv_delay:
-            prev_height = self.adb.get_tx_height(prev_txid)
-            if prev_height.height > 0:
-                wanted_height = prev_height.height + sweep_info.csv_delay - 1
-            else:
-                wanted_height = local_height + sweep_info.csv_delay
-            if wanted_height - local_height > 0:
-                can_broadcast = False
-                # self.logger.debug(
-                #     f"pending redeem for {prevout}. waiting for {name}: CSV "
-                #     f"({local_height=}, {wanted_height=}, {prev_height.height=}, {sweep_info.csv_delay=})")
-        if can_broadcast:
-            self.logger.info(f'we can broadcast: {name}')
-            if await self.network.try_broadcasting(new_tx, name):
-                tx_was_added = self.adb.add_transaction(new_tx, is_new=(old_tx is None))
-            else:
-                tx_was_added = False
-        else:
-            # we may have a tx with a different fee, in which case it will be replaced
-            if not old_tx or (old_tx and old_tx.txid() != new_tx.txid()):
-                try:
-                    tx_was_added = self.adb.add_transaction(new_tx, is_new=(old_tx is None))
-                except Exception as e:
-                    self.logger.info(f'could not add future tx: {name}. prevout: {prevout} {str(e)}')
-                    tx_was_added = False
-                if tx_was_added:
-                    self.logger.info(f'added redeem tx: {name}. prevout: {prevout}')
-            else:
-                tx_was_added = False
-            # set future tx regardless of tx_was_added, because it is not persisted
-            # (and wanted_height can change if input of CSV was not mined before)
-            self.adb.set_future_tx(new_tx.txid(), wanted_height=wanted_height)
-        if tx_was_added:
-            self.lnworker.wallet.set_label(new_tx.txid(), name)
-            if old_tx and old_tx.txid() != new_tx.txid():
-                self.lnworker.wallet.set_label(old_tx.txid(), None)
-            util.trigger_callback('wallet_updated', self.lnworker.wallet)
+    def maybe_redeem(self, sweep_info: 'SweepInfo') -> bool:
+        """ returns False if it was dust """
+        try:
+            self.lnworker.wallet.txbatcher.add_sweep_input('lnwatcher', sweep_info, self.config.FEE_POLICY_LIGHTNING)
+        except BelowDustLimit:
+            return False
+        return True
 
     def maybe_extract_preimage(self, chan: 'AbstractChannel', spender_tx: Transaction, prevout: str):
         txin_idx = spender_tx.get_input_idx_that_spent_prevout(TxOutpoint.from_str(prevout))

electrum/simple_config.py

@@ -675,7 +675,9 @@ def __setattr__(self, name, value):
     TEST_SHUTDOWN_FEE_RANGE = ConfigVar('test_shutdown_fee_range', default=None)
     TEST_SHUTDOWN_LEGACY = ConfigVar('test_shutdown_legacy', default=False, type_=bool)
 
-    FEE_POLICY = ConfigVar('fee_policy', default='eta:2', type_=str)
+    FEE_POLICY = ConfigVar('fee_policy', default='eta:2', type_=str) # exposed to GUI
+    FEE_POLICY_LIGHTNING = ConfigVar('fee_policy_lightning', default='eta:2', type_=str) # for txbatcher (sweeping)
+    FEE_POLICY_SWAPS = ConfigVar('fee_policy_swaps', default='eta:2', type_=str) # for txbatcher (sweeping and sending if we are a swapserver)
 
     RPC_USERNAME = ConfigVar('rpcuser', default=None, type_=str)
     RPC_PASSWORD = ConfigVar('rpcpassword', default=None, type_=str)