From 815b345aea634580ba9b1a1c3680f98233b17cad Mon Sep 17 00:00:00 2001 From: f321x Date: Wed, 19 Aug 2026 15:24:25 +0200 Subject: [PATCH 1/3] interface: ensure subscription queue order is increasing This ensures the order of responses on the subscription queue is either duplicated or increasing in recency but not decreasing. Before it could happen that `handle_request` already receives a newer notification and puts it on the subscription queue, and then `send_response` returning the initial response and also putting it on the queue. Causing the less recent `send_response` result to pop from the queue after the newer one. The following synchronizer commit depends on the messages arriving in order, otherwise it might drop a newer status in favor of an older one that was put on the queue after the newer one. --- electrum/interface.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/electrum/interface.py b/electrum/interface.py index 1b862d2cc41..bc999518840 100644 --- a/electrum/interface.py +++ b/electrum/interface.py @@ -235,14 +235,14 @@ async def subscribe(self, method: str, params: List, queue: asyncio.Queue): # note: multiple Synchronizers (from different Wallet objects) might sub to the same key, # hence subscriptions map key->list[queue] self.subscriptions[key].append(queue) - if key in self.subs_cache: - result = self.subs_cache[key] - else: + if key not in self.subs_cache: # note: until subs_cache is written for the first time, # each 'subscribe' call might make a request on the network. result = await self.send_request(method, params) - self.subs_cache[key] = result - await queue.put(params + [result]) + # don't override what was already set in handle_request, it might be newer than the send_request response + if key not in self.subs_cache: + self.subs_cache[key] = result + await queue.put(params + [self.subs_cache[key]]) def unsubscribe(self, queue): """Unsubscribe a callback to free object references to enable GC.""" From f13a38899223fc4bb9a6f0e9e0cda00b66e62624 Mon Sep 17 00:00:00 2001 From: f321x Date: Tue, 18 Aug 2026 16:53:50 +0200 Subject: [PATCH 2/3] synchronizer: fix race during concurrent status updates We would incorrectly disconnect from a server when this race occurs: 1. We receive S1 (e.g. tx in mempool) 2. We request history H1 for S1 -> req1 3. We receive S2 (e.g. tx got mined) concurrently 4. We request history H2 for S2 -> req2 5. We receive H2 for S2 (before req1 got satisfied) 6. req1 receives H2 for current status (S2) from the server as well 7. _on_address_status schedules disconnect because H2 mismatches S1 8. S2 got already handled before, we never cancel the disconnect ``` DEBUG electrum.tests.toyserver.toyserver.ToyServerSession:toyserver.py:465 127.0.0.1:42170 disconnected ERROR electrum.interface.Interface.[127.0.0.1:46059]:util.py:1231 Exception in run: SynchronizerFailure('timeout reached waiting for addr bcrt1qyt2w5jw5ktkr0ta96zuque93l4p86u3k2tj36cxcvs6gzl6wyj5q04p7y3: history still stale') Traceback (most recent call last): File "/home/runner/work/electrum/electrum/electrum/util.py", line 1225, in wrapper return await func(*args, **kwargs) File "/home/runner/work/electrum/electrum/electrum/interface.py", line 756, in wrapper_func return await func(self, *args, **kwargs) File "/home/runner/work/electrum/electrum/electrum/interface.py", line 782, in run await self.open_session(ssl_context=ssl_context) File "/home/runner/work/electrum/electrum/electrum/interface.py", line 1072, in open_session async with self.taskgroup as group: File "/opt/hostedtoolcache/Python/3.10.20/x64/lib/python3.10/site-packages/aiorpcx/curio.py", line 304, in __aexit__ await self.join() File "/home/runner/work/electrum/electrum/electrum/util.py", line 1430, in join task.result() File "/home/runner/work/electrum/electrum/electrum/util.py", line 1564, in run_tasks_wrapper await self._run_tasks(taskgroup=taskgroup) File "/home/runner/work/electrum/electrum/electrum/synchronizer.py", line 79, in _run_tasks async with taskgroup as group: File "/opt/hostedtoolcache/Python/3.10.20/x64/lib/python3.10/site-packages/aiorpcx/curio.py", line 304, in __aexit__ await self.join() File "/home/runner/work/electrum/electrum/electrum/util.py", line 1430, in join task.result() File "/home/runner/work/electrum/electrum/electrum/synchronizer.py", line 216, in disconnect_if_still_stale raise SynchronizerFailure(f"timeout reached waiting for addr {addr}: history still stale") electrum.synchronizer.SynchronizerFailure: timeout reached waiting for addr bcrt1qyt2w5jw5ktkr0ta96zuque93l4p86u3k2tj36cxcvs6gzl6wyj5q04p7y3: history still stale ``` --- electrum/synchronizer.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/electrum/synchronizer.py b/electrum/synchronizer.py index 7da384df8ec..c74f0786ab0 100644 --- a/electrum/synchronizer.py +++ b/electrum/synchronizer.py @@ -68,6 +68,7 @@ def _reset(self): self._adding_addrs = set() self.requested_addrs = set() self._handling_addr_statuses = set() + self._last_announced_status = {} # type: Dict[str, Optional[str]] self.scripthash_to_address = {} self._processed_some_notifications = False # so that we don't miss them # Queues @@ -124,6 +125,7 @@ async def handle_status(self): assert_hash256_str(status) # process status addr = self.scripthash_to_address[sh] + self._last_announced_status[addr] = status self._handling_addr_statuses.add(addr) self.requested_addrs.discard(addr) # ok for addr not to be present await self.taskgroup.spawn(self._on_address_status, addr, status) @@ -201,11 +203,11 @@ async def _on_address_status(self, addr, status): self._handling_addr_statuses.discard(addr) result = await self._maybe_request_history_for_addr(addr, ann_status=status) hist = list(map(lambda item: (item['tx_hash'], item['height']), result)) - # tx_fees - tx_fees = [(item['tx_hash'], item.get('fee')) for item in result] - tx_fees = dict(filter(lambda x:x[1] is not None, tx_fees)) + if status != self._last_announced_status.get(addr): + # The server already sent us a newer status while we have been waiting for this history response. + self.logger.debug(f"discarding obsolete history for {addr}") # Check that the status corresponds to what was announced - if history_status(hist) != status: + elif history_status(hist) != status: # could happen naturally if history changed between getting status and history (race) self.logger.info(f"error: status mismatch: {addr}. we'll wait a bit for status update.") # The server is supposed to send a new status notification, which will trigger a new @@ -217,6 +219,9 @@ async def disconnect_if_still_stale(): self._stale_histories[addr] = await self.taskgroup.spawn(disconnect_if_still_stale) else: self._stale_histories.pop(addr, asyncio.Future()).cancel() + # tx_fees + tx_fees = [(item['tx_hash'], item.get('fee')) for item in result] + tx_fees = dict(filter(lambda x: x[1] is not None, tx_fees)) # Store received history self.adb.receive_history_callback(addr, hist, tx_fees) # Request transactions we don't have From 8d8187094ed34c098bb5888eb177b72f446c5528 Mon Sep 17 00:00:00 2001 From: f321x Date: Tue, 18 Aug 2026 17:12:52 +0200 Subject: [PATCH 3/3] synchronizer: cancel stale history task if status matches our history If the status reverts to the one our stored history already has, _on_address_status returns early and the task disconnect stays active, so we drop a server that behaved correctly. Also cancel previous disconnect task before scheduling to prevent the following (rather unlikely) scenario: 1. We receive S1 -> req1 2. We receive S2 -> req2 3. req2 answers mismatching S2 -> disconnect task (_stale_histories[addr] = DT2) 4. We receive S1 again, req1 is still in flight, so (addr, S1) is still in requested_histories -> _on_address_status returns early and does not cancel DT2 5. req1 answers mismatching S1, S1 is the current status, so this is not obsolete -> we arm DT1 and assign _stale_histories[addr] = DT1, dropping DT2 6. A later status with matching history cancels _stale_histories[addr], which is DT1. DT2 is unreachable, fires, and disconnects a server that behaved correctly. --- electrum/synchronizer.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/electrum/synchronizer.py b/electrum/synchronizer.py index c74f0786ab0..c764a75f564 100644 --- a/electrum/synchronizer.py +++ b/electrum/synchronizer.py @@ -191,6 +191,7 @@ async def _on_address_status(self, addr, status): try: old_history = self.adb.db.get_addr_history(addr) if history_status(old_history) == status: + self._stale_histories.pop(addr, asyncio.Future()).cancel() return # No point in requesting history twice for the same announced status. # However if we got announced a new status, we should request history again: @@ -216,6 +217,7 @@ async def disconnect_if_still_stale(): timeout = self.network.get_network_timeout_seconds(NetworkTimeout.Generic) await asyncio.sleep(timeout) raise SynchronizerFailure(f"timeout reached waiting for addr {addr}: history still stale") + self._stale_histories.pop(addr, asyncio.Future()).cancel() self._stale_histories[addr] = await self.taskgroup.spawn(disconnect_if_still_stale) else: self._stale_histories.pop(addr, asyncio.Future()).cancel()