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.""" diff --git a/electrum/synchronizer.py b/electrum/synchronizer.py index 7da384df8ec..c764a75f564 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) @@ -189,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: @@ -201,11 +204,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 @@ -214,9 +217,13 @@ 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() + # 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