synchronizer: outpoint subs: always send spk_hint to server

Author: SomberNight

electrum/address_synchronizer.py

@@ -102,7 +102,7 @@ def __init__(self, db: 'WalletDB', config: 'SimpleConfig', *, name: str = None):
 
         self._get_balance_cache = {}
         self._get_utxos_cache = {}
-        self._subscribed_outpoints = set()
+        self._subscribed_outpoints = {}  # type: dict[str, str]  # outpoint->spk_hint
 
         self.load_and_cleanup()
 
@@ -239,13 +239,13 @@ def add_address(self, address: str) -> None:
             self.synchronizer.add_address(address)
         self.up_to_date_changed()
 
-    def subscribe_to_outpoint(self, outpoint):
+    def subscribe_to_outpoint(self, outpoint: str, *, spk_hint: str) -> None:
         if outpoint in self._subscribed_outpoints:
             return
-        self._subscribed_outpoints.add(outpoint)
+        self._subscribed_outpoints[outpoint] = spk_hint
         if self.synchronizer:
             # fixme: launch polling task in synchronizer
-            self.synchronizer.add_outpoint(outpoint)
+            self.synchronizer.add_outpoint(outpoint, spk_hint=spk_hint)
         self.up_to_date_changed()
 
     @with_lock
@@ -1079,7 +1079,7 @@ def subscribe_to_tx_outpoints(self, spender_txid: str):
         spender_tx = self.get_transaction(spender_txid)
         for i, o in enumerate(spender_tx.outputs()):
             txo = spender_txid + ':%d'%i
-            self.subscribe_to_outpoint(txo)
+            self.subscribe_to_outpoint(txo, spk_hint=o.scriptpubkey.hex())
 
     def get_tx_mined_depth(self, txid: str):
         if not txid:

electrum/interface.py

@@ -238,17 +238,21 @@ def set_default_timeout(self, timeout):
     async def subscribe(self, method: str, params: List, queue: asyncio.Queue):
         # note: until the cache is written for the first time,
         # each 'subscribe' call might make a request on the network.
-        params2 = params[:]
+        notif_params = params[:]
         if method == 'blockchain.scriptpubkey.subscribe':
-            params2[0] = script_to_scripthash(bytes.fromhex(params2[0]))
-        key = self.get_hashable_key_for_rpc_call(method, params2)
+            # params: spk
+            notif_params[0] = script_to_scripthash(bytes.fromhex(params[0]))  # convert spk->sh
+        elif method == 'blockchain.outpoint.subscribe':
+            # params: tx_hash, txout_idx, spk_hint
+            notif_params = notif_params[0:2]  # drop spk_hint
+        key = self.get_hashable_key_for_rpc_call(method, notif_params)
         self.subscriptions[key].append(queue)
         if key in self.cache:
             result = self.cache[key]
         else:
             result = await self.send_request(method, params)
             self.cache[key] = result
-        await queue.put(params2 + [result])
+        await queue.put(notif_params + [result])
 
     def unsubscribe(self, queue):
         """Unsubscribe a callback to free object references to enable GC."""

electrum/lnchannel.py

@@ -36,6 +36,7 @@
 from electrum_ecc import ECPubkey
 
 from . import constants, util
+from . import bitcoin
 from .util import bfh, chunks, TxMinedInfo, error_text_bytes_to_safe_str
 from .bitcoin import redeem_script_to_address
 from .crypto import sha256, sha256d
@@ -481,6 +482,9 @@ def is_funding_tx_mined(self, funding_height: TxMinedInfo) -> bool:
     def get_funding_address(self) -> str:
         pass
 
+    def get_funding_scriptpubkey(self) -> bytes:
+        return bitcoin.address_to_script(self.get_funding_address())
+
     def get_funding_tx(self) -> Optional[Transaction]:
         funding_txid = self.funding_outpoint.txid
         return self.lnworker.lnwatcher.adb.get_transaction(funding_txid)

electrum/lnwatcher.py

@@ -26,7 +26,7 @@ def __init__(self, lnworker: 'LNWallet'):
         Logger.__init__(self)
         self.adb = lnworker.wallet.adb
         self.config = lnworker.config
-        self.callbacks = {}  # type: Dict[str, Callable[[], Awaitable[None]]]  # address -> lambda function
+        self.callbacks = {}  # type: Dict[str, Callable[[], Awaitable[None]]]  # address/txo -> lambda function
         self.network = None
         self.register_callbacks()
         self._pending_force_closes = set()
@@ -62,10 +62,11 @@ def add_outpoint_callback(
         outpoint: str,
         callback: Callable[[], Awaitable[None]],
         *,
+        spk_hint: str,
         subscribe: bool = True,
     ) -> None:
         if subscribe:
-            self.adb.subscribe_to_outpoint(outpoint)
+            self.adb.subscribe_to_outpoint(outpoint, spk_hint=spk_hint)
         self.callbacks[outpoint] = callback
 
     async def trigger_callbacks(self, *, requires_synchronizer: bool = True):
@@ -107,7 +108,11 @@ async def on_event_adb_set_up_to_date(self, adb):
     def add_channel(self, chan: 'AbstractChannel') -> None:
         outpoint = chan.funding_outpoint.to_str()
         callback = lambda: self.check_onchain_situation(outpoint)
-        self.add_outpoint_callback(outpoint, callback, subscribe=chan.need_to_subscribe())
+        self.add_outpoint_callback(
+            outpoint, callback,
+            spk_hint=chan.get_funding_scriptpubkey().hex(),
+            subscribe=chan.need_to_subscribe(),
+        )
 
     @ignore_exceptions
     @log_exceptions
@@ -121,7 +126,7 @@ async def check_onchain_situation(self, funding_outpoint: str) -> None:
         closing_txid = self.adb.get_spender(funding_outpoint)
         closing_height = self.adb.get_tx_height(closing_txid)
         if closing_txid:
-            self.adb.subscribe_to_tx_outpoints(closing_txid)
+            self.adb.subscribe_to_tx_outpoints(closing_txid)  # FIXME this assumes we alrdy have closing_tx
             closing_tx = self.adb.get_transaction(closing_txid)
             if closing_tx:
                 keep_watching = await self.sweep_commitment_transaction(funding_outpoint, closing_tx)

electrum/synchronizer.py

@@ -66,7 +66,7 @@ def __init__(self, network: 'Network'):
     def _reset(self):
         super()._reset()
         self._adding_addrs = set()
-        self._adding_outpoints = set()
+        self._adding_outpoints = {}  # type: dict[str, str]  # outpoint->spk_hint
         self.requested_addrs = set()
         self.requested_outpoints = set()
         self._handling_addr_statuses = set()
@@ -93,8 +93,9 @@ def add_address(self, addr: str) -> None:
         if not is_address(addr): raise ValueError(f"invalid bitcoin address {neuter_bitcoin_address(addr)}")
         self._adding_addrs.add(addr)  # this lets is_up_to_date already know about addr
 
-    def add_outpoint(self, outpoint: str) -> None:
-        self._adding_outpoints.add(outpoint)  # this lets is_up_to_date already know about outpoint
+    def add_outpoint(self, outpoint: str, *, spk_hint: str) -> None:
+        # this lets is_up_to_date already know about outpoint
+        self._adding_outpoints[outpoint] = spk_hint
 
     async def _add_address(self, addr: str):
         try:
@@ -105,13 +106,13 @@ async def _add_address(self, addr: str):
         finally:
             self._adding_addrs.discard(addr)  # ok for addr not to be present
 
-    async def _add_outpoint(self, outpoint: str):
+    async def _add_outpoint(self, outpoint: str, *, spk_hint: str):
         try:
             if outpoint in self.requested_outpoints: return
             self.requested_outpoints.add(outpoint)
-            await self.taskgroup.spawn(self._subscribe_to_outpoint, outpoint)
+            await self.taskgroup.spawn(self._subscribe_to_outpoint(outpoint, spk_hint=spk_hint))
         finally:
-            self._adding_outpoints.discard(outpoint)  # ok for addr not to be present
+            self._adding_outpoints.pop(outpoint, None)  # ok for outpoint not to be present
 
     async def _on_address_status(self, addr: str, status: Optional[str]):
         """Handle the change of the status of an address.
@@ -137,14 +138,14 @@ async def _subscribe_to_address(self, addr):
             raise
         self._requests_answered += 1
 
-    async def _subscribe_to_outpoint(self, outpoint):
+    async def _subscribe_to_outpoint(self, outpoint: str, *, spk_hint: str) -> None:
         self._requests_sent += 1
         txhash, idx = outpoint.split(':')
         idx = int(idx)
         self.logger.info(f'subscribe to outpoint: {txhash}:{idx}')
         try:
             async with self._network_request_semaphore:
-                await self.session.subscribe('blockchain.outpoint.subscribe', [txhash, idx], self.outpoint_status_queue)
+                await self.session.subscribe('blockchain.outpoint.subscribe', [txhash, idx, spk_hint], self.outpoint_status_queue)
         except RPCError as e:
             raise
         self._requests_answered += 1
@@ -365,17 +366,17 @@ async def main(self):
         for addr in random_shuffled_copy(self.adb.get_addresses()):
             await self._add_address(addr)
         # add outpoints to bootstrap (race)
-        for outpoint in self.adb._subscribed_outpoints:
-            await self._add_outpoint(outpoint)
+        for outpoint, spk_hint in self.adb._subscribed_outpoints.items():
+            await self._add_outpoint(outpoint, spk_hint=spk_hint)
         # main loop
         self._init_done = True
         prev_uptodate = False
         while True:
             await asyncio.sleep(0.1)
             for addr in self._adding_addrs.copy(): # copy set to ensure iterator stability
                 await self._add_address(addr)
-            for outpoint in list(self._adding_outpoints):
-                await self._add_outpoint(outpoint)
+            for outpoint, spk_hint in self._adding_outpoints.copy().items():
+                await self._add_outpoint(outpoint, spk_hint=spk_hint)
             up_to_date = self.adb.is_up_to_date()
             # see if status changed
             if (up_to_date != prev_uptodate