txbatcher: support multiple batches, with different fee policies

Author: ecdsa

electrum/gui/qt/confirm_tx_dialog.py

@@ -30,7 +30,7 @@
 from PyQt6.QtCore import Qt
 from PyQt6.QtGui import QIcon
 
-from PyQt6.QtWidgets import QHBoxLayout, QVBoxLayout, QLabel, QGridLayout, QPushButton, QToolButton, QMenu
+from PyQt6.QtWidgets import QHBoxLayout, QVBoxLayout, QLabel, QGridLayout, QPushButton, QToolButton, QMenu, QComboBox
 
 from electrum.i18n import _
 from electrum.util import NotEnoughFunds, NoDynamicFeeEstimates
@@ -79,6 +79,8 @@ def __init__(self, *, title='',
         # preview is disabled for lightning channel funding
         self._allow_preview = allow_preview
         self.is_preview = False
+        self._base_tx = None # for batching
+        self.batching_candidates = self.wallet.get_unconfirmed_base_tx_for_batching([], []) # todo: update it
 
         self.locktime_e = LockTimeEdit(self)
         self.locktime_e.valueEdited.connect(self.trigger_update)
@@ -112,7 +114,7 @@ def __init__(self, *, title='',
         self.main_window.gui_object.timer.timeout.connect(self.timer_actions)
 
     def is_batching(self):
-        return self.config.WALLET_BATCH_RBF and not self.main_window.utxo_list.is_coincontrol_active()
+        return self._base_tx and not self.main_window.utxo_list.is_coincontrol_active()
 
     def should_show_io(self):
         return self.config.GUI_QT_TX_EDITOR_SHOW_IO
@@ -151,6 +153,12 @@ def update_fee_target(self):
     def update_feerate_label(self):
         self.feerate_label.setText(self.feerate_e.text() + ' ' + self.feerate_e.base_unit())
 
+    def update_fee_controls(self):
+        self.fee_slider.setEnabled(not self.is_batching())
+        self.fee_combo.setEnabled(not self.is_batching())
+        self.fee_slider.setVisible(not self.is_batching())
+        self.fee_combo.setVisible(not self.is_batching())
+
     def create_fee_controls(self):
 
         self.fee_label = QLabel('')
@@ -380,6 +388,15 @@ def create_buttons_bar(self):
         self.ok_button.clicked.connect(self.on_send)
         self.ok_button.setDefault(True)
         buttons = Buttons(CancelButton(self), self.preview_button, self.ok_button)
+
+        if self.batching_candidates:
+            batching_combo = QComboBox()
+            batching_combo.addItems([_('Not batching')] + [_('Batch with') + ' ' + x.txid()[0:10] for x in self.batching_candidates])
+            buttons.insertWidget(0, batching_combo)
+            def on_batching_combo(x):
+                self._base_tx = self.batching_candidates[x - 1] if x > 0 else None
+                self.update_batching()
+            batching_combo.currentIndexChanged.connect(on_batching_combo)
         return buttons
 
     def create_top_bar(self, text):
@@ -419,7 +436,6 @@ def add_cv_action(configvar: 'ConfigVarWithConfig', action: Callable[[], None]):
                 _('This may result in higher transactions fees.')
             ]))
         self.use_multi_change_menu.setEnabled(self.wallet.use_change)
-        add_cv_action(self.config.cv.WALLET_BATCH_RBF, self.toggle_batch_rbf)
         add_cv_action(self.config.cv.WALLET_MERGE_DUPLICATE_OUTPUTS, self.toggle_merge_duplicate_outputs)
         add_cv_action(self.config.cv.WALLET_SPEND_CONFIRMED_ONLY, self.toggle_confirmed_only)
         add_cv_action(self.config.cv.WALLET_COIN_CHOOSER_OUTPUT_ROUNDING, self.toggle_output_rounding)
@@ -456,9 +472,8 @@ def toggle_multiple_change(self):
         self.wallet.db.put('multiple_change', self.wallet.multiple_change)
         self.trigger_update()
 
-    def toggle_batch_rbf(self):
-        b = not self.config.WALLET_BATCH_RBF
-        self.config.WALLET_BATCH_RBF = b
+    def update_batching(self):
+        self.update_fee_controls()
         self.set_io_visible()
         self.resize_to_fit_content()
         self.trigger_update()
@@ -581,7 +596,7 @@ def get_messages(self):
             messages.append(_('This transaction will spend unconfirmed coins.'))
         # warn if we merge from mempool
         if self.is_batching():
-            messages.append(_('Transaction batching is active. The fee will be bumped automatically if needed'))
+            messages.append(_('Transaction batching is active. The fee policy of the selected batch will be used'))
         else:
             # warn if we use multiple change outputs
             num_change = sum(int(o.is_change) for o in self.tx.outputs())
@@ -652,17 +667,17 @@ def _update_amount_label(self):
     def update_tx(self, *, fallback_to_zero_fee: bool = False):
         fee_estimator = self.get_fee_estimator()
         confirmed_only = self.config.WALLET_SPEND_CONFIRMED_ONLY
-        is_batching = self.is_batching()
+        base_tx = self._base_tx
         try:
-            self.tx = self.make_tx(fee_estimator, confirmed_only=confirmed_only, is_batching=is_batching)
+            self.tx = self.make_tx(fee_estimator, confirmed_only=confirmed_only, base_tx=base_tx)
             self.not_enough_funds = False
             self.no_dynfee_estimates = False
         except NotEnoughFunds:
             self.not_enough_funds = True
             self.tx = None
             if fallback_to_zero_fee:
                 try:
-                    self.tx = self.make_tx(0, confirmed_only=confirmed_only, is_batching=is_batching)
+                    self.tx = self.make_tx(0, confirmed_only=confirmed_only, base_tx=base_tx)
                 except BaseException:
                     return
             else:
@@ -671,7 +686,7 @@ def update_tx(self, *, fallback_to_zero_fee: bool = False):
             self.no_dynfee_estimates = True
             self.tx = None
             try:
-                self.tx = self.make_tx(0, confirmed_only=confirmed_only, is_batching=is_batching)
+                self.tx = self.make_tx(0, confirmed_only=confirmed_only, base_tx=base_tx)
             except NotEnoughFunds:
                 self.not_enough_funds = True
                 return
@@ -686,7 +701,7 @@ def update_tx(self, *, fallback_to_zero_fee: bool = False):
     def can_pay_assuming_zero_fees(self, confirmed_only) -> bool:
         # called in send_tab.py
         try:
-            tx = self.make_tx(0, confirmed_only=confirmed_only, is_batching=False)
+            tx = self.make_tx(0, confirmed_only=confirmed_only, base_tx=None)
         except NotEnoughFunds:
             return False
         else:
@@ -709,7 +724,7 @@ def create_grid(self):
         grid.addWidget(HelpLabel(_("Mining Fee") + ": ", msg), 1, 0)
         grid.addLayout(self.fee_hbox, 1, 1, 1, 3)
 
-        grid.addWidget(HelpLabel(_("Fee target") + ": ", self.fee_combo.help_msg), 3, 0)
+        grid.addWidget(HelpLabel(_("Fee policy") + ": ", self.fee_combo.help_msg), 3, 0)
         grid.addLayout(self.fee_target_hbox, 3, 1, 1, 3)
 
         grid.setColumnStretch(4, 1)

electrum/gui/qt/main_window.py

@@ -1422,7 +1422,7 @@ def confirm_tx_dialog(self, make_tx, output_value, allow_preview=True):
                 text = self.send_tab.get_text_not_enough_funds_mentioning_frozen()
                 self.show_message(text)
                 return
-        return d.run(), d.is_preview, d.is_batching()
+        return d.run(), d.is_preview, d._base_tx
 
     @protected
     def _open_channel(self, connect_str, funding_sat, push_amt, funding_tx, password):

electrum/gui/qt/send_tab.py

@@ -326,9 +326,8 @@ def pay_onchain_dialog(
         if get_coins is None:
             get_coins = self.window.get_coins
 
-        def make_tx(fee_est, *, confirmed_only=False, is_batching=False):
-            base_tx = self.wallet.txbatcher.get_base_tx() if is_batching else None
-            merge_duplicate_outputs = self.config.WALLET_MERGE_DUPLICATE_OUTPUTS if not is_batching else False
+        def make_tx(fee_est, *, confirmed_only=False, base_tx=None):
+            merge_duplicate_outputs = self.config.WALLET_MERGE_DUPLICATE_OUTPUTS if not base_tx else False
             coins = get_coins(nonlocal_only=nonlocal_only, confirmed_only=confirmed_only)
             return self.wallet.make_unsigned_transaction(
                 coins=coins,
@@ -344,7 +343,7 @@ def make_tx(fee_est, *, confirmed_only=False, is_batching=False):
         is_max = any(parse_max_spend(outval) for outval in output_values)
         output_value = '!' if is_max else sum(output_values)
 
-        tx, is_preview, is_batching = self.window.confirm_tx_dialog(make_tx, output_value)
+        tx, is_preview, base_tx = self.window.confirm_tx_dialog(make_tx, output_value)
         if tx is None:
             # user cancelled
             return
@@ -361,7 +360,7 @@ def make_tx(fee_est, *, confirmed_only=False, is_batching=False):
                     self.show_error(str(e))
                     return
 
-                if is_batching:
+                if base_tx:
                     self.save_pending_invoice()
                     funding_output = PartialTxOutput.from_address_and_value(swap.lockup_address, swap_dummy_output.value)
                     coro = sm.wait_for_htlcs_and_broadcast(transport, swap=swap, invoice=invoice, output=funding_output)
@@ -378,10 +377,11 @@ def make_tx(fee_est, *, confirmed_only=False, is_batching=False):
                     tx.swap_invoice = invoice
                     tx.swap_payment_hash = swap.payment_hash
 
-        if is_batching:
+        if base_tx:
             self.save_pending_invoice()
+            key = self.wallet.txbatcher.find_key_of_txid(tx.txid())
             for output in outputs:
-                self.wallet.txbatcher.add_batch_payment(output)
+                self.wallet.txbatcher.add_batch_payment(output, key)
             return
 
         if is_preview:

electrum/lnwatcher.py

@@ -286,4 +286,4 @@ async def maybe_redeem(self, sweep_info: 'SweepInfo') -> None:
             # careful, this prevents revocation as well
             if not self.lnworker.enable_htlc_settle_onchain:
                 return
-        self.lnworker.wallet.txbatcher.add_sweep_info(sweep_info)
+        self.lnworker.wallet.txbatcher.add_sweep_info(sweep_info, 'default')

electrum/submarine_swaps.py

@@ -417,7 +417,7 @@ async def _claim_swap(self, swap: SwapData) -> None:
                 txout=None,
                 name='swap claim',
             )
-            self.wallet.txbatcher.add_sweep_info(sweep_info)
+            self.wallet.txbatcher.add_sweep_info(sweep_info, 'default')
 
     def get_swap_tx_fee(self):
         return self.get_fee(SWAP_TX_SIZE)
@@ -450,7 +450,7 @@ async def hold_invoice_callback(self, payment_hash: bytes) -> None:
             swap = self.swaps[key]
             if not swap.is_funded():
                 output = self.create_funding_output(swap)
-                self.wallet.txbatcher.add_batch_payment(output)
+                self.wallet.txbatcher.add_batch_payment(output, 'default')
                 swap._payment_pending = True
         else:
             self.logger.info(f'key not in swaps {key}')
@@ -737,7 +737,7 @@ async def callback(payment_hash):
             if tx:
                 await self.broadcast_funding_tx(swap, tx)
             else:
-                self.wallet.txbatcher.add_batch_payment(output)
+                self.wallet.txbatcher.add_batch_payment(output, 'default')
 
         self.lnworker.register_hold_invoice(payment_hash, callback)
 

electrum/txbatcher.py

@@ -70,26 +70,80 @@
 # In order to generalize that logic to payments, callers would need to pass a unique ID along with
 # the payment output, so that we can prevent paying twice.
 
-from .json_db import locked
+from .json_db import locked  # todo: move to util.py
+
+
+class FeePolicy:
+    # todo: fee slider should have an instance of this object instead of side-effecting config
+    def estimate_fee(self, size: int, allow_fallback_to_static_rates=True):
+        return 42
 
 class TxBatcher(Logger):
 
     SLEEP_INTERVAL = 1
-    RETRY_DELAY = 60
 
     def __init__(self, wallet):
+        self.batch_txids = wallet.db.get_stored_item("tx_batches", { })
+        self.txbatchers = {}
+        self.wallet = wallet
+        for key in self.batch_txids.keys():
+            self._maybe_create_batch(key, policy=FeePolicy())
+
+    def _maybe_create_batch(self, key, policy):
+        if key not in self.txbatchers:
+            self.txbatchers[key] = TxBatch(self.wallet, policy, self.batch_txids, key)
+
+    def add_batch_payment(self, output: 'PartialTxOutput', key):
+        self._maybe_create_batch(key, policy=FeePolicy())
+        self.txbatchers[key].add_batch_payment(output)
+
+    def add_sweep_info(self, sweep_info: 'SweepInfo', key):
+        self._maybe_create_batch(key, policy=FeePolicy())
+        self.txbatchers[key].add_sweep_info(sweep_info)
+
+    def find_batch_of_txid(self, txid) -> str:
+        # used by gui
+        for k, v in self.batch_txids.items():
+            if txid in v:
+                return k
+        return 'batch:' + txid
+
+    def is_mine(self, txid):
+        for k, v in self.batch_txids.items():
+            if txid in v:
+                return True
+        return False
+
+    @log_exceptions
+    async def run(self):
+        while True:
+            await asyncio.sleep(self.SLEEP_INTERVAL)
+            password = self.wallet.get_unlocked_password()
+            if self.wallet.has_keystore_encryption() and not password:
+                continue
+            for txbatcher in self.txbatchers.values():
+                await txbatcher.run_iteration(password)
+
+class TxBatch(Logger):
+
+    RETRY_DELAY = 60
+
+    def __init__(self, wallet, fee_policy, batch_txids, key):
         Logger.__init__(self)
         self.wallet = wallet
-        self.config = wallet.config
+        self.key = key
+        self.fee_policy = fee_policy
         self.lock = threading.RLock()
         self.batch_payments = []       # list of payments we need to make
         self.batch_inputs = {}         # list of inputs we need to sweep
         # list of tx that were broadcast. Each tx is a RBF replacement of the previous one. Ony one can get mined.
-        self._batch_txids = wallet.db.get_stored_item("batch_txids", [])
+        self._batch_txids = batch_txids # fixme: check persistence
+        if self.key not in self._batch_txids:
+            self._batch_txids[self.key] = []
         self._base_tx = None           # current batch tx. last element of batch_txids
 
-        if self._batch_txids:
-            last_txid = self._batch_txids[-1]
+        if self._batch_txids[self.key]:
+            last_txid = self._batch_txids[self.key][-1]
             tx = self.wallet.adb.get_transaction(last_txid)
             if tx:
                 tx = PartialTransaction.from_tx(tx)
@@ -101,7 +155,7 @@ def __init__(self, wallet):
         self._unconfirmed_sweeps = set()  # list of inputs we are sweeping (until spending tx is confirmed)
 
     def is_mine(self, txid):
-        return txid in self._batch_txids
+        return txid in self._batch_txids[self.key]
 
     @locked
     def add_batch_payment(self, output: 'PartialTxOutput'):
@@ -139,7 +193,7 @@ def get_base_tx(self) -> Optional[Transaction]:
         return self._base_tx
 
     def _find_confirmed_base_tx(self) -> Optional[Transaction]:
-        for txid in self._batch_txids:
+        for txid in self._batch_txids[self.key]:
             tx_mined_status = self.wallet.adb.get_tx_height(txid)
             if tx_mined_status.conf > 0:
                 tx = self.wallet.adb.get_transaction(txid)
@@ -184,19 +238,14 @@ def _should_bump_fee(self, base_tx) -> bool:
         if not self.is_mine(base_tx.txid()):
             return False
         base_tx_fee = base_tx.get_fee()
-        recommended_fee = self.config.estimate_fee(base_tx.estimated_size(), allow_fallback_to_static_rates=True)
+        recommended_fee = self.fee_policy.estimate_fee(base_tx.estimated_size(), allow_fallback_to_static_rates=True)
         should_bump_fee = base_tx_fee * 1.1 < recommended_fee
         if should_bump_fee:
             self.logger.info(f'base tx fee too low {base_tx_fee} < {recommended_fee}. we will bump the fee')
         return should_bump_fee
 
-    @log_exceptions
-    async def run(self):
-        while True:
-            await asyncio.sleep(self.SLEEP_INTERVAL)
-            password = self.wallet.get_unlocked_password()
-            if self.wallet.has_keystore_encryption() and not password:
-                continue
+
+    async def run_iteration(self, password):
             await self._maybe_broadcast_legacy_htlc_txs()
             tx = self._find_confirmed_base_tx()
             if tx:
@@ -214,22 +263,24 @@ async def run(self):
                 else:
                     self.wallet.add_future_tx(v, wanted_height)
             if not to_pay and not to_sweep_now and not self._should_bump_fee(base_tx):
-                continue
+                return
             try:
                 tx = self._create_batch_tx(base_tx, to_sweep_now, to_pay, password)
             except Exception as e:
                 self.logger.exception(f'Cannot create batch transaction: {repr(e)}')
                 if base_tx:
                     self._start_new_batch(base_tx)
-                    continue
-                await asyncio.sleep(self.RETRY_DELAY)
-                continue
+                    return
+                else:
+                    raise
+                #await asyncio.sleep(self.RETRY_DELAY)
+                #return
             self.logger.info(f'created tx with {len(tx.inputs())} inputs and {len(tx.outputs())} outputs')
             self.logger.info(f'{str(tx)}')
             if await self.wallet.network.try_broadcasting(tx, 'batch'):
                 self.wallet.adb.add_transaction(tx)
                 if tx.has_change():
-                    self._batch_txids.append(tx.txid())
+                    self._batch_txids[self.key].append(tx.txid())
                     self._base_tx = tx
                 else:
                     self.logger.info(f'starting new batch because current base tx does not have change')
@@ -288,7 +339,7 @@ def _start_new_batch(self, tx):
         use_change = tx and tx.has_change() and any([txout in self.batch_payments for txout in tx.outputs()])
         self.batch_payments = self._to_pay_after(tx)
         self.batch_inputs = self._to_sweep_after(tx)
-        self._batch_txids.clear()
+        self._batch_txids[self.key] = []
         self._base_tx = None
         self._parent_tx = tx if use_change else None