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/address_synchronizer.py

@@ -360,7 +360,7 @@ def add_value_from_prev_output():
             self.db.add_transaction(tx_hash, tx)
             self.db.add_num_inputs_to_tx(tx_hash, len(tx.inputs()))
             if is_new:
-                util.trigger_callback('adb_added_tx', self, tx_hash, tx)
+                util.trigger_callback('adb_added_tx', self, tx_hash, tx, conflicting_txns)
             return True
 
     def remove_transaction(self, tx_hash: str) -> None:

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,9 @@ 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=True,
+            )
         else:
             # second-stage
             address = bitcoin.script_to_p2wsh(htlctx_witness_script)
@@ -404,6 +411,7 @@ def txs_htlc(
                         cltv_abs=0,
                         txin=sweep_txin,
                         txout=None,
+                        can_be_batched=False, # todo: we could batch htlcs of the same channel
                     )
 
     # offered HTLCs, in our ctx --> "timeout"
@@ -541,6 +549,7 @@ def sweep_their_ctx_to_remote_backup(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
 
     # to_remote
@@ -562,6 +571,7 @@ def sweep_their_ctx_to_remote_backup(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
     return txs
 
@@ -619,6 +629,7 @@ def sweep_their_ctx(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=True,
             )
 
     # to_local is handled by lnwatcher
@@ -631,6 +642,7 @@ def sweep_their_ctx(
                 cltv_abs=None,
                 txin=txin,
                 txout=None,
+                can_be_batched=False,
             )
 
     # to_remote
@@ -662,6 +674,7 @@ def sweep_their_ctx(
                     cltv_abs=None,
                     txin=txin,
                     txout=None,
+                    can_be_batched=True,
                 )
 
     # HTLCs
@@ -701,6 +714,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

@@ -80,6 +80,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:
@@ -146,6 +153,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)
@@ -241,9 +252,8 @@ 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):
@@ -253,7 +263,7 @@ async def sweep_commitment_transaction(self, funding_outpoint, closing_tx) -> bo
             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
@@ -267,7 +277,7 @@ async def sweep_commitment_transaction(self, funding_outpoint, closing_tx) -> bo
                         keep_watching |= not self.is_deeply_mined(htlc_tx_spender)
                     else:
                         keep_watching = True
-                    await self.maybe_redeem(prevout2, htlc_sweep_info, name)
+                        await 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))
@@ -276,111 +286,31 @@ async def sweep_commitment_transaction(self, funding_outpoint, closing_tx) -> bo
                 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)
-
+                # broadcast or maybe update our own tx
+                await 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=})")
+    async def maybe_redeem(self, sweep_info: 'SweepInfo') -> None:
         if not (sweep_info.cltv_abs or sweep_info.csv_delay):
             # used to control settling of htlcs onchain for testing purposes
             # careful, this prevents revocation as well
             if not self.lnworker.enable_htlc_settle_onchain:
                 return
-        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)
+        # pre-anchor htlc txs cannot be batched
+        if sweep_info.name == 'first-stage-htlc':
+            await self._maybe_redeem_legacy_htlc_tx(sweep_info)
+            return
+        # if it cannot be batched, we still send it to txbatcher, so that the fee will be bumped
+        batch_key = 'lnwatcher' if sweep_info.can_be_batched else sweep_info.txin.prevout.to_str()
+        self.lnworker.wallet.txbatcher.add_sweep_input(batch_key, sweep_info, self.fee_policy)
+
+    async def _maybe_redeem_legacy_htlc_tx(self, sweep_info):
+        local_height = self.network.get_local_height()
+        wanted_height = sweep_info.cltv_abs
+        if wanted_height - local_height > 0:
+            return
+        self.logger.info('legacy first-stage htlc tx')
+        tx = PartialTransaction.from_io([sweep_info.txin], [sweep_info.txout], locktime=sweep_info.cltv_abs, version=2)
+        self.lnworker.wallet.sign_transaction(tx, password=None, ignore_warnings=True)
+        if await self.network.try_broadcasting(tx, sweep_info.name):
+            self.lnworker.wallet.adb.add_transaction(tx)