Merge pull request #10877 from f321x/swaps_pop_invoice

### electrum/lnworker.py
@@ -1063,7 +1063,7 @@ def __init__(self, wallet: 'Abstract_Wallet', xprv, *, features: LnFeatures = No
         self._channel_sending_capacity_lock = asyncio.Lock()
 
         # detect inflight payments
-        self.inflight_payments = set()        # (not persisted) keys of invoices that are in PR_INFLIGHT state
+        self.inflight_payments = set()  # type: set[str]  # (not persisted) keys of invoices that are in PR_INFLIGHT state
         for payment_hash in self.get_payments(status='inflight').keys():
             self.set_invoice_status(payment_hash.hex(), PR_INFLIGHT)
 
@@ -1907,7 +1907,7 @@ async def pay_invoice(
         invoice_features = lnaddr.get_features()
         r_tags = lnaddr.get_routing_info('r')
         amount_to_pay = lnaddr.get_amount_msat()
-        status = self.get_payment_status(payment_hash, direction=SENT)
+        status = self.get_invoice_status(invoice)
         if status == PR_PAID:
             raise PaymentFailure(_("This invoice has been paid already"))
         if status == PR_INFLIGHT:
@@ -1931,7 +1931,7 @@ async def pay_invoice(
         if attempts is None and self.uses_trampoline():
             # we don't expect lots of failed htlcs with trampoline, so we can fail sooner
             attempts = 30
-        success = False
+        success, reason = False, _("unknown")
         try:
             await self.pay_to_node(
                 node_pubkey=invoice_pubkey,
@@ -1947,20 +1947,17 @@ async def pay_invoice(
                 budget=budget,
             )
             success = True
-        except PaymentFailure as e:
-            self.logger.info(f'payment failure: {e!r}')
-            reason = str(e)
-        except ChannelDBNotLoaded as e:
+        except (PaymentFailure, ChannelDBNotLoaded) as e:
             self.logger.info(f'payment failure: {e!r}')
             reason = str(e)
         finally:
             self.logger.info(f"pay_invoice ending session for RHASH={payment_hash.hex()}. {success=}")
-        if success:
-            self.set_invoice_status(key, PR_PAID)
-            util.trigger_callback('payment_succeeded', self.wallet, key)
-        else:
-            self.set_invoice_status(key, PR_UNPAID)
-            util.trigger_callback('payment_failed', self.wallet, key, reason)
+            if success:
+                self.set_invoice_status(key, PR_PAID)
+                util.trigger_callback('payment_succeeded', self.wallet, key)
+            else:
+                self.set_invoice_status(key, PR_UNPAID)  # allows retries (unless there are still unresolved htlcs)
+                util.trigger_callback('payment_failed', self.wallet, key, reason)
         log = self.logs[key]
         return success, log
 

### electrum/plugins/swapserver/server.py
@@ -122,7 +122,7 @@ async def get_pairs(self, r):
 
     async def add_swap_invoice(self, r):
         request = await r.json()
-        self.sm.server_add_swap_invoice(request)
+        await self.sm.server_add_swap_invoice(request)
         return web.json_response({})
 
     async def create_normal_swap(self, r):

### electrum/submarine_swaps.py
@@ -38,8 +38,7 @@
     get_nostr_ann_pow_amount, make_aiohttp_proxy_connector, get_running_loop, get_asyncio_loop, wait_for2,
     run_sync_function_on_asyncio_thread, trigger_callback, NoDynamicFeeEstimates, UserFacingException, now
 )
-from . import lnutil
-from .lnutil import hex_to_bytes, Keypair
+from .lnutil import hex_to_bytes, Keypair, SENT, RECEIVED, MIN_FINAL_CLTV_DELTA_ACCEPTED, PaymentFailure
 from .bolt11 import decode_bolt11_invoice
 from .stored_dict import StoredObject, stored_at
 from . import constants
@@ -87,7 +86,7 @@
 assert SPENDER_FINALITY_DELAY < MIN_LOCKTIME_DELTA_FOR_CLAIM
 assert MIN_LOCKTIME_DELTA_FOR_CLAIM < MIN_LOCKTIME_DELTA
 assert MIN_LOCKTIME_DELTA <= LOCKTIME_DELTA_REFUND <= MAX_LOCKTIME_DELTA
-assert MAX_LOCKTIME_DELTA + SPENDER_FINALITY_DELAY < lnutil.MIN_FINAL_CLTV_DELTA_ACCEPTED
+assert MAX_LOCKTIME_DELTA + SPENDER_FINALITY_DELAY < MIN_FINAL_CLTV_DELTA_ACCEPTED
 assert MAX_LOCKTIME_DELTA + SPENDER_FINALITY_DELAY < MIN_FINAL_CLTV_DELTA_FOR_CLIENT
 
 
@@ -283,7 +282,7 @@ def __init__(self, *, wallet: 'Abstract_Wallet', lnworker: 'LNWallet'):
                 self._prepayments[swap.prepay_hash] = payment_hash
             if not swap.is_reverse and not swap.is_redeemed and not self.lnworker.get_preimage(swap.payment_hash):
                 if (swap.prepay_hash is not None
-                        and self.lnworker.get_payment_status(swap.prepay_hash, direction=lnutil.RECEIVED) != PR_PAID):
+                        and self.lnworker.get_payment_status(swap.prepay_hash, direction=RECEIVED) != PR_PAID):
                     # re-bundle payments, because lnworker does not persist bundles.
                     # note: if the prepay is already PR_PAID, lnpeer completed and deleted the bundle before
                     #       shutdown; re-creating it would make is_payment_bundle_complete() permanently False.
@@ -347,17 +346,14 @@ async def run_nostr_server(self):
 
     @log_exceptions
     async def main_loop(self):
-        tasks = [self.pay_pending_invoices()]
-        if self.is_server:
-            # nostr and http are not mutually exclusive
-            if self.config.SWAPSERVER_PORT:
-                tasks.append(self.http_server.run())
-            if self.config.NOSTR_RELAYS:
-                tasks.append(self.run_nostr_server())
-
+        # nostr and http are not mutually exclusive
         async with self.taskgroup as group:
-            for task in tasks:
-                await group.spawn(task)
+            if self.is_server:
+                if self.config.SWAPSERVER_PORT:
+                    await group.spawn(self.http_server.run())
+                if self.config.NOSTR_RELAYS:
+                    await group.spawn(self.run_nostr_server())
+            await group.spawn(asyncio.Event().wait())  # run until cancel
 
     async def stop(self):
         await self.taskgroup.cancel_remaining()
@@ -453,32 +449,6 @@ async def set_nostr_proof_of_work(self) -> None:
         assert pow_amount >= self.config.SWAPSERVER_POW_TARGET, pow_amount
         self.config.SWAPSERVER_ANN_POW_NONCE = nonce
 
-    async def pay_invoice(self, key):
-        self.logger.info(f'trying to pay invoice {key}')
-        self.invoices_to_pay[key] = 1000000000000 # lock
-        try:
-            invoice = self.wallet.get_invoice(key)
-            success, log = await self.lnworker.pay_invoice(invoice)
-        except Exception as e:
-            self.logger.info(f'exception paying {key}, will not retry')
-            self.invoices_to_pay.pop(key, None)
-            return
-        if not success:
-            self.logger.info(f'failed to pay {key}, will retry in 10 minutes')
-            self.invoices_to_pay[key] = now() + 600
-        else:
-            self.logger.info(f'paid invoice {key}')
-            self.invoices_to_pay.pop(key, None)
-
-    async def pay_pending_invoices(self):
-        self.invoices_to_pay = {}
-        while True:
-            await asyncio.sleep(5)
-            for key, not_before in list(self.invoices_to_pay.items()):
-                if now() < not_before:
-                    continue
-                await self.taskgroup.spawn(self.pay_invoice(key))
-
     def cancel_normal_swap(self, swap: Optional[SwapData], *, reason: str = 'user cancelled') -> bool:
         """Fail/cancel the swap, unless its funding tx is already being broadcast. Safe to call from the GUI thread."""
         if swap is None:
@@ -492,32 +462,34 @@ def cancel_normal_swap(self, swap: Optional[SwapData], *, reason: str = 'user ca
         return True
 
     def _fail_swap(self, swap: SwapData, reason: str):
-        self.logger.info(f'failing swap {swap.payment_hash.hex()}: {reason}')
         swap._is_cancelled = True
-        if not swap.is_reverse and swap.payment_hash in self.lnworker.hold_invoice_callbacks:
+        key = swap.payment_hash.hex()
+        lnw, wallet, lnwatcher = self.lnworker, self.wallet, self.lnwatcher
+        self.logger.info(f'failing swap {key}: {reason}')
+        if not swap.is_reverse and swap.payment_hash in lnw.hold_invoice_callbacks:
             # unregister_hold_invoice will fail pending htlcs if there is no preimage available
-            self.lnworker.unregister_hold_invoice(swap.payment_hash)
-            self.lnworker.delete_payment_info(swap.payment_hash.hex(), direction=lnutil.RECEIVED)
-            self.lnworker.clear_invoices_cache()
+            lnw.unregister_hold_invoice(swap.payment_hash)
+            lnw.delete_payment_info(key, direction=RECEIVED)
+            lnw.clear_invoices_cache()
         if not swap.is_funded():
-            self.lnwatcher.remove_callback(swap.lockup_address)
+            lnwatcher.remove_callback(swap.lockup_address)
             with self.swaps_lock:
-                swaps = self.wallet.db.get_dict('submarine_swaps')
-                if swaps.pop(swap.payment_hash.hex(), None) is None:
-                    self.logger.debug(f"swap {swap.payment_hash.hex()} has already been deleted.")
-                self._swaps.pop(swap.payment_hash.hex(), None)
+                swaps = wallet.db.get_dict('submarine_swaps')
+                if swaps.pop(key, None) is None:
+                    self.logger.debug(f"swap {key} has already been deleted.")
+                self._swaps.pop(key, None)
                 if swap._funding_prevout is not None:
                     self._swaps_by_funding_outpoint.pop(swap._funding_prevout, None)
                 self._swaps_by_lockup_address.pop(swap.lockup_address, None)
                 if swap.prepay_hash is not None:
                     self._prepayments.pop(swap.prepay_hash, None)
-                    if self.lnworker.get_payment_status(swap.prepay_hash, direction=lnutil.RECEIVED) != PR_PAID:
-                        self.lnworker.delete_payment_info(swap.prepay_hash.hex(), direction=lnutil.RECEIVED)
-                        self.lnworker.delete_payment_bundle(payment_hash=swap.payment_hash)
-                    if self.lnworker.get_payment_status(swap.prepay_hash, direction=lnutil.SENT) != PR_PAID:
-                        self.lnworker.delete_payment_info(swap.prepay_hash.hex(), direction=lnutil.SENT)
-                if self.lnworker.get_payment_status(swap.payment_hash, direction=lnutil.SENT) != PR_PAID:
-                    self.lnworker.delete_payment_info(swap.payment_hash.hex(), direction=lnutil.SENT)
+                    if lnw.get_payment_status(swap.prepay_hash, direction=RECEIVED) != PR_PAID:
+                        lnw.delete_payment_info(swap.prepay_hash.hex(), direction=RECEIVED)
+                        lnw.delete_payment_bundle(payment_hash=swap.payment_hash)
+                    if lnw.get_payment_status(swap.prepay_hash, direction=SENT) != PR_PAID:
+                        lnw.delete_payment_info(swap.prepay_hash.hex(), direction=SENT)
+                if lnw.get_payment_status(swap.payment_hash, direction=SENT) != PR_PAID:
+                    lnw.delete_payment_info(key, direction=SENT)
 
     def _get_public_preimage(self, swap: SwapData) -> Optional[bytes]:
         if swap.spending_txid is None:
@@ -839,10 +811,10 @@ def add_normal_swap(
         self.lnworker.add_payment_info_for_hold_invoice(
             payment_hash,
             lightning_amount_sat=invoice_amount_sat,
-            min_final_cltv_delta=min_final_cltv_expiry_delta or lnutil.MIN_FINAL_CLTV_DELTA_ACCEPTED,
+            min_final_cltv_delta=min_final_cltv_expiry_delta or MIN_FINAL_CLTV_DELTA_ACCEPTED,
             exp_delay=300,
         )
-        info = self.lnworker.get_payment_info(payment_hash, direction=lnutil.RECEIVED)
+        info = self.lnworker.get_payment_info(payment_hash, direction=RECEIVED)
         lnaddr1, invoice = self.lnworker.get_bolt11_invoice(
             payment_info=info,
             message='Submarine swap',
@@ -858,10 +830,10 @@ def add_normal_swap(
         if prepay:
             prepay_hash = self.lnworker.create_payment_info(
                 amount_msat=prepay_amount_sat*1000,
-                min_final_cltv_delta=min_final_cltv_expiry_delta or lnutil.MIN_FINAL_CLTV_DELTA_ACCEPTED,
+                min_final_cltv_delta=min_final_cltv_expiry_delta or MIN_FINAL_CLTV_DELTA_ACCEPTED,
                 exp_delay=300,
             )
-            info = self.lnworker.get_payment_info(prepay_hash, direction=lnutil.RECEIVED)
+            info = self.lnworker.get_payment_info(prepay_hash, direction=RECEIVED)
             lnaddr2, prepay_invoice = self.lnworker.get_bolt11_invoice(
                 payment_info=info,
                 message='Submarine swap prepayment',
@@ -970,7 +942,7 @@ def add_reverse_swap(
         self.add_lnwatcher_callback(swap)
         return swap
 
-    def server_add_swap_invoice(self, request: dict) -> dict:
+    async def server_add_swap_invoice(self, request: dict) -> dict:
         """ server method.
         (client-forward-swap phase2)
         """
@@ -994,10 +966,7 @@ def server_add_swap_invoice(self, request: dict) -> dict:
                 payment_hash=payment_hash, locktime=swap.locktime, refund_pubkey=their_pubkey, claim_pubkey=our_pubkey,
             )
             assert swap.redeem_script == redeem_script
-            assert key not in self.invoices_to_pay
-            self.invoices_to_pay[key] = 0
-            assert self.wallet.get_invoice(invoice.get_id()) is None
-            self.wallet.save_invoice(invoice)
+        await self.taskgroup.spawn(self.pay_invoice_safe(invoice))
         return {}
 
     async def normal_swap(
@@ -1741,6 +1710,16 @@ def get_pending_swaps(self) -> List[SwapData]:
                 pending_swaps.append(swap)
         return pending_swaps
 
+    async def pay_invoice_safe(self, invoice: 'Invoice'):
+        try:
+            success, htlc_log = await self.lnworker.pay_invoice(invoice)  # prevents duplicate payment attempts
+            if not success:
+                raise PaymentFailure("\n".join(str(x.formatted_tuple()) for x in htlc_log))
+        except PaymentFailure as e:
+            self.logger.warning(f"failed to pay swap invoice {invoice.get_id()}: {e}")
+        except Exception:
+            self.logger.exception("exception while paying swap invoice")
+
 
 class SwapServerTransport(Logger):
 
@@ -2200,7 +2179,7 @@ async def _handle_requests(self) -> None:
                 method = request.pop('method')
                 self.logger.info(f'handle_request: id={event_id} {method} {request}')
                 if method == 'addswapinvoice':  # client-forward-swap phase2
-                    r = self.sm.server_add_swap_invoice(request)
+                    r = await self.sm.server_add_swap_invoice(request)
                 elif method == 'createswap':  # client-reverse-swap
                     r = self.sm.server_create_swap(request)
                 elif method == 'createnormalswap':  # client-forward-swap phase1

### tests/test_lnwallet.py
@@ -10,12 +10,13 @@
 from electrum import bitcoin
 import electrum.trampoline
 from electrum.channel_db import UpdateStatus
-from electrum.lnutil import RECEIVED, MIN_FINAL_CLTV_DELTA_ACCEPTED, serialize_htlc_key, LnFeatures, HTLCOwner
+from electrum.lnutil import RECEIVED, SENT, MIN_FINAL_CLTV_DELTA_ACCEPTED, serialize_htlc_key, LnFeatures, HTLCOwner, PaymentFailure
 from electrum.logging import console_stderr_handler
 from electrum.lnmsg import decode_msg
 from electrum.lnrouter import RouteEdge
+from electrum.bolt11 import encode_bolt11_invoice, BOLT11Addr
 from electrum.lntransport import LNPeerAddr
-from electrum.invoices import LN_EXPIRY_NEVER, PR_UNPAID
+from electrum.invoices import LN_EXPIRY_NEVER, PR_UNPAID, PR_INFLIGHT, Invoice
 from electrum.lnpeer import Peer
 from electrum.lnchannel import Channel, ChannelState
 from electrum.lnonion import OnionPacket, OnionRoutingFailure, OnionFailureCode
@@ -73,6 +74,39 @@ def test_create_payment_info__amount_must_not_be_zero(self):
                 exp_delay=exp_delay,
             )
 
+    async def test_pay_invoice_rejects_second_attempt_while_first_is_inflight(self):
+        """A second attempt to pay an invoice with equal payment hash must be rejected while an earlier attempt is
+        still running, even before that attempt has committed any htlc to a channel yet.
+        """
+        sender = self.lnwallet_anchors
+        recipient = self.create_mock_lnwallet(name='recipient')
+        lnaddr, _pay_req = lnhelpers.prepare_invoice(recipient)
+        payment_hash = lnaddr.paymenthash
+        key = payment_hash.hex()
+        # same payment hash, but different payment secret
+        pay_req2 = Invoice.from_bech32(encode_bolt11_invoice(
+            BOLT11Addr(
+                paymenthash=payment_hash,
+                amount=lnaddr.amount,
+                tags=[
+                    ('c', MIN_FINAL_CLTV_DELTA_ACCEPTED),
+                    ('d', 'test'),
+                    ('9', recipient.features.for_bolt11_invoice()),
+                    ('x', 3600),
+                ],
+                payment_secret=os.urandom(32),
+            ),
+            recipient.node_keypair.privkey))
+        self.assertEqual(key, pay_req2.rhash)
+
+        # first payment attempt sets invoice status inflight but no htlcs have been added yet (e.g. during pathfinding)
+        sender.set_invoice_status(key, PR_INFLIGHT)
+        self.assertEqual({}, sender.get_payments(status='inflight'))
+
+        with self.assertRaises(PaymentFailure):
+            await sender.pay_invoice(pay_req2)
+        self.assertNotIn(key, sender.logs)  # rejected before pay_to_node opened a session
+
     async def test_trampoline_invoice_features_and_routing_hints(self):
         """
         When the invoice_features signal trampoline support, routing hints must only