synchronizer: fix race during concurrent status updates

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."""

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