batch payment manager:

The class TxBatcher handles the creation, broadcast and replacement
of replaceable transactions. Callers (LNWatcher, SwapManager) use
methods add_payment_output and add_sweep_info. Transactions
created by TxBatcher may combine sweeps and outgoing payments.

Transactions created by TxBatcher will have their fee bumped
automatically (this was only the case for sweeps before).

TxBatcher manages several TxBatches. TxBatches are created
dynamically when needed.

The GUI does not touch txbatcher transactions:
  - wallet.get_candidates_for_batching excludes txbatcher
    transactions
  - RBF dialogs do not work with txbatcher transactions

wallet:
  - instead of reading config variables, make_unsigned_transaction
    takes new parameters: base_tx, send_change_to_lighting

tests:
  - unit tests in test_txbatcher.py (replaces test_sswaps.py)
  - force all regtests to use MPP, so that we sweep transactions
    with several HTLCs. This forces the payment manager to aggregate
    first-stage HTLC tx inputs. second-stage are not batched for now.

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,7 @@ def txs_htlc(
                         cltv_abs=0,
                         txin=sweep_txin,
                         txout=None,
+                        can_be_batched=True, # this is safe, we are the only ones who can spend (assuming we did not broadcast a revoked state)
                     )
 
     # offered HTLCs, in our ctx --> "timeout"
@@ -541,6 +551,7 @@ def sweep_their_ctx_to_remote_backup(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
 
     # to_remote
@@ -562,6 +573,7 @@ def sweep_their_ctx_to_remote_backup(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
     return txs
 
@@ -619,6 +631,7 @@ def sweep_their_ctx(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
 
     # to_local is handled by lnwatcher
@@ -631,6 +644,7 @@ def sweep_their_ctx(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=False,
             )
 
     # to_remote
@@ -656,12 +670,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 +717,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,19 +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
 
 
@@ -75,6 +70,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 +143,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)
@@ -212,18 +218,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
@@ -236,19 +230,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
@@ -261,116 +252,21 @@ 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)
                 # extract preimage
                 keep_watching |= not self.is_deeply_mined(spender_txid)
                 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)
             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 whether it was added """
+        try:
+            self.lnworker.wallet.txbatcher.add_sweep_input('lnwatcher', sweep_info, self.fee_policy)
+        except BelowDustLimit:
+            return False
+        return True