wallet: share prevout lookups within one paid-detection pass

_is_onchain_invoice_paid redoes the prevout lookup and per-prevout
get_tx_height for each invoice, even though invoices often share a
scriptpubkey (repeat payments to the same destination). One
_update_onchain_invoice_paid_detection pass over such invoices does
the identical lookup for each of them, and unconfirmed invoices
(conf==0, e.g. during initial sync before merkle proofs arrive) cannot
enter the paid-keys cache, so during sync every tx addition pays the
full cost again, on the asyncio event loop thread.

Memoize the per-scriptpubkey lookup for the duration of one detection
pass; the per-invoice height filtering still applies on top. The cache
only exists while a pass runs, so it needs no invalidation: adb state
cannot change mid-pass, as both mutate on the asyncio thread. It also
covers the second lookup that get_invoice_status does for invoices
that cannot be cache-seeded (unconfirmed/unpaid).

On a real wallet (686 invoices, 572 sharing one destination):
- detection pass at wallet open: ~18s -> ~0.6s (~36s on master)
- one adb_added_tx event while invoices are unconfirmed (initial
  sync): ~18s -> ~0.6s (~35s on master)
A full sync from seed with these invoices present went from 12+ hours
(GUI frozen, servers timing out) to ~5 minutes.

Author: sashazykov

electrum/wallet.py

@@ -423,6 +423,8 @@ def __init__(self, db: WalletDB, *, config: SimpleConfig):
         self._last_full_history = None
         self._tx_parents_cache = {}
         self._paid_invoice_keys_cache = set()  # type: Set[str]
+        # only set while an _update_onchain_invoice_paid_detection pass runs, see there
+        self._invoice_prevouts_cache = None  # type: Optional[Dict[bytes, Sequence[Tuple[str, int, int, int]]]]
         self._coin_price_cache = {}
         self._default_labels = {}
         self._accounting_addresses = set()  # addresses counted as ours after successful sweep
@@ -1376,33 +1378,39 @@ def _prepare_onchain_invoice_paid_detection(self):
         self._update_onchain_invoice_paid_detection(self._invoices.keys())
 
     def _update_onchain_invoice_paid_detection(self, invoice_keys: Iterable[str]) -> None:
-        for invoice_key in invoice_keys:
-            invoice = self._invoices.get(invoice_key)
-            if not invoice:
-                continue
-            # clear the cache first so self.get_invoice_status takes the slow path,
-            # which is needed to detect paid->unpaid transitions (e.g. reorgs)
-            self._paid_invoice_keys_cache.discard(invoice_key)
-            if invoice.is_lightning():
-                is_paid_lightning = bool(self.lnworker and self.lnworker.get_invoice_status(invoice) == PR_PAID)
-                if is_paid_lightning:
-                    self._paid_invoice_keys_cache.add(invoice_key)
-                if is_paid_lightning or not invoice.get_address():
+        # invoices often share scriptpubkeys; share the prevout lookups across this pass,
+        # including the ones done by self.get_invoice_status below
+        self._invoice_prevouts_cache = {}
+        try:
+            for invoice_key in invoice_keys:
+                invoice = self._invoices.get(invoice_key)
+                if not invoice:
                     continue
-            is_paid, conf_needed, relevant_txs = self._is_onchain_invoice_paid(invoice)
-            if is_paid:
-                for txid in relevant_txs:
-                    self._invoices_from_txid_map[txid].add(invoice_key)
-                # re-add to the cache, so that self.get_invoice_status below short-circuits
-                # instead of redoing the prevout scan. not for lightning invoices:
-                # their LN status takes precedence there (e.g. an in-flight LN payment)
-                if not invoice.is_lightning() and conf_needed:
-                    self._paid_invoice_keys_cache.add(invoice_key)
-            for txout in invoice.get_outputs():
-                self._invoices_from_scriptpubkey_map[txout.scriptpubkey].add(invoice_key)
-            # update invoice status
-            status = self.get_invoice_status(invoice)
-            util.trigger_callback('invoice_status', self, invoice_key, status)
+                # clear the cache first so self.get_invoice_status takes the slow path,
+                # which is needed to detect paid->unpaid transitions (e.g. reorgs)
+                self._paid_invoice_keys_cache.discard(invoice_key)
+                if invoice.is_lightning():
+                    is_paid_lightning = bool(self.lnworker and self.lnworker.get_invoice_status(invoice) == PR_PAID)
+                    if is_paid_lightning:
+                        self._paid_invoice_keys_cache.add(invoice_key)
+                    if is_paid_lightning or not invoice.get_address():
+                        continue
+                is_paid, conf_needed, relevant_txs = self._is_onchain_invoice_paid(invoice)
+                if is_paid:
+                    for txid in relevant_txs:
+                        self._invoices_from_txid_map[txid].add(invoice_key)
+                    # re-add to the cache, so that self.get_invoice_status below short-circuits
+                    # instead of redoing the prevout scan. not for lightning invoices:
+                    # their LN status takes precedence there (e.g. an in-flight LN payment)
+                    if not invoice.is_lightning() and conf_needed:
+                        self._paid_invoice_keys_cache.add(invoice_key)
+                for txout in invoice.get_outputs():
+                    self._invoices_from_scriptpubkey_map[txout.scriptpubkey].add(invoice_key)
+                # update invoice status
+                status = self.get_invoice_status(invoice)
+                util.trigger_callback('invoice_status', self, invoice_key, status)
+        finally:
+            self._invoice_prevouts_cache = None
 
     def _is_onchain_invoice_paid(self, invoice: BaseInvoice) -> Tuple[bool, Optional[int], Sequence[str]]:
         """Returns whether on-chain invoice/request is satisfied, num confs required txs have,
@@ -1419,15 +1427,12 @@ def _is_onchain_invoice_paid(self, invoice: BaseInvoice) -> Tuple[bool, Optional
         conf_needed = None  # type: Optional[int]
         with self.lock:
             for invoice_scriptpubkey, invoice_amt in invoice_amounts.items():
-                scripthash = bitcoin.script_to_scripthash(invoice_scriptpubkey)
-                prevouts_and_values = self.db.get_prevouts_by_scripthash(scripthash)
                 confs_and_values = []
-                for prevout, v in prevouts_and_values:
-                    relevant_txs.add(prevout.txid.hex())
-                    tx_height = self.adb.get_tx_height(prevout.txid.hex())
-                    if 0 < tx_height.height() <= invoice.height:  # exclude txs older than invoice
+                for txid, height, conf, v in self._get_prevout_infos_for_scriptpubkey(invoice_scriptpubkey):
+                    relevant_txs.add(txid)
+                    if 0 < height <= invoice.height:  # exclude txs older than invoice
                         continue
-                    confs_and_values.append((tx_height.conf or 0, v))
+                    confs_and_values.append((conf, v))
                 # check that there is at least one TXO, and that they pay enough.
                 # note: "at least one TXO" check is needed for zero amount invoice (e.g. OP_RETURN)
                 vsum = 0
@@ -1440,6 +1445,23 @@ def _is_onchain_invoice_paid(self, invoice: BaseInvoice) -> Tuple[bool, Optional
                     is_paid = False
         return is_paid, conf_needed, list(relevant_txs)
 
+    def _get_prevout_infos_for_scriptpubkey(self, scriptpubkey: bytes) -> Sequence[Tuple[str, int, int, int]]:
+        """Returns (txid, height, conf, value) for each prevout paying scriptpubkey.
+        Memoized while an _update_onchain_invoice_paid_detection pass runs (see there).
+        """
+        cache = self._invoice_prevouts_cache
+        infos = cache.get(scriptpubkey) if cache is not None else None
+        if infos is None:
+            scripthash = bitcoin.script_to_scripthash(scriptpubkey)
+            infos = []
+            for prevout, v in self.db.get_prevouts_by_scripthash(scripthash):
+                txid = prevout.txid.hex()
+                tx_height = self.adb.get_tx_height(txid)
+                infos.append((txid, tx_height.height(), tx_height.conf or 0, v))
+            if cache is not None:
+                cache[scriptpubkey] = infos
+        return infos
+
     def is_onchain_invoice_paid(self, invoice: BaseInvoice) -> Tuple[bool, Optional[int]]:
         is_paid, conf_needed, relevant_txs = self._is_onchain_invoice_paid(invoice)
         return is_paid, conf_needed

tests/test_invoices.py

@@ -275,15 +275,15 @@ def _make_wallet(self):
         wallet.adb.receive_tx_callback(funding_tx, tx_height=TX_HEIGHT_UNCONFIRMED)
         return wallet
 
-    def _make_outgoing_invoice(self, dest_addr: str, amount_sat: int, *, t: int = 1700000000):
+    def _make_outgoing_invoice(self, dest_addr: str, amount_sat: int, *, t: int = 1700000000, height: int = 0):
         outputs = [PartialTxOutput.from_address_and_value(dest_addr, amount_sat)]
         return Invoice(
             amount_msat=amount_sat * 1000,
             message="",
             time=t,
             exp=LN_EXPIRY_NEVER,
             outputs=outputs,
-            height=0,
+            height=height,
             lightning_invoice=None,
         )
 
@@ -505,6 +505,44 @@ def spy(invoice):
                          "tx verification must not rescan an invoice already cached as PR_PAID")
         self.assertEqual(PR_PAID, wallet.get_invoice_status(inv))
 
+    async def test_detection_pass_looks_up_each_scripthash_once(self):
+        """Invoices sharing a destination scriptpubkey must not each redo the same
+        prevout lookup within one detection pass; the per-invoice height filtering
+        must still apply on top of the shared lookup."""
+        wallet = self._make_wallet()
+        dest = "tb1qmjzmg8nd4z56ar4fpngzsr6euktrhnjg9td385"
+        invs = [self._make_outgoing_invoice(dest, 1_000 + i, t=1700000000 + i) for i in range(5)]
+        # This one was created above the paying tx's height, so the tx must not count for it.
+        inv_late = self._make_outgoing_invoice(dest, 1_000, t=1700000010, height=1005)
+        for inv in invs + [inv_late]:
+            wallet.save_invoice(inv, write_to_disk=False)
+        # Pay the shared destination (confirmed at height 1001).
+        outputs = [PartialTxOutput.from_address_and_value(dest, 50_000)]
+        tx = wallet.make_unsigned_transaction(outputs=outputs, fee_policy=FixedFeePolicy(1000))
+        wallet.sign_transaction(tx, password=None)
+        wallet.adb.receive_tx_callback(tx, tx_height=TX_HEIGHT_UNCONFIRMED)
+        wallet.db.put('stored_height', 1010)
+        wallet.adb.add_verified_tx(tx.txid(), TxMinedInfo(_height=1001, timestamp=1700000001, txpos=1, header_hash="01"*32))
+
+        # Track the prevout lookups.
+        looked_up = []
+        orig = wallet.db.get_prevouts_by_scripthash
+        def spy(scripthash):
+            looked_up.append(scripthash)
+            return orig(scripthash)
+        wallet.db.get_prevouts_by_scripthash = spy
+
+        wallet._paid_invoice_keys_cache.clear()  # force the slow path
+        wallet._prepare_onchain_invoice_paid_detection()
+
+        self.assertEqual(1, len(looked_up),
+                         "one shared scriptpubkey must be looked up exactly once per pass")
+        # The shared lookup must not change the per-invoice outcomes.
+        for inv in invs:
+            self.assertIn(inv.get_id(), wallet._paid_invoice_keys_cache)
+        self.assertNotIn(inv_late.get_id(), wallet._paid_invoice_keys_cache)
+        self.assertEqual(PR_UNPAID, wallet.get_invoice_status(inv_late))
+
     async def test_paid_keys_demoted_on_reorg(self):
         """A reorg unverifying the paying tx must remove the invoice from the cache,
         otherwise get_invoice_status would keep returning a stale PR_PAID."""