Corroborate an external spend before marking a UTXO gone (B-29)

detect_external_spends marked a UTXO spent_txid='external-spend'
irreversibly the moment it was missing from one listunspent reply, on
one server, with no way to undo it. A rotated-to server that's broken
or behind, or an empty reply, could zero a user's balance permanently.

Split into three functions in deposits/service.py: find_utxos_missing_
from (read-only candidate detection, and refuses to flag anything at
all when listunspent comes back entirely empty for a funded address -
that reads as a broken response, not a full sweep), mark_utxos_spent_
externally (persistence only, once a candidate is already confirmed),
and reinstate_reappeared_utxos (undoes the mark if the outpoint
reappears as unspent later).

ElectrumListener gains corroborate_utxo_spent, sharing the same
majority-quorum logic corroborate_header already uses for B-28: before
a candidate is marked, the other configured servers are asked whether
they also see it as spent. _refresh_user now reads candidates, then
corroborates each one with no DB session held open across those
network calls (same shape as B-18/B-25), then persists.

Suite grows from 165 to 176 tests. BUGS.md moves B-29 to Previously
fixed.

app/electrum/listener.py

@@ -6,7 +6,12 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import async_sessionmaker
 
 from app.db.models import User
-from app.deposits.service import credit_confirmed_utxos, detect_external_spends
+from app.deposits.service import (
+    credit_confirmed_utxos,
+    find_utxos_missing_from,
+    mark_utxos_spent_externally,
+    reinstate_reappeared_utxos,
+)
 from app.electrum.client import ElectrumClient, ElectrumEndpoint
 from app.electrum.scripthash import address_to_scripthash
 from app.rounds.draw import (
@@ -24,10 +29,10 @@ logger = logging.getLogger(__name__)
 # difference between noticing the drop in a minute and never noticing it at all.
 _PING_INTERVAL_SECONDS = 60
 
-# B-28: how long to wait for any *one* other server's answer when corroborating the
-# draw's block header. Shorter than the standard request timeout since this is a
-# supplementary check across several servers at once — a single slow fallback
-# shouldn't hold up the others.
+# How long to wait for any *one* other server's answer when corroborating the
+# draw's block header (B-28) or a candidate external spend (B-29). Shorter than
+# the standard request timeout since this is a supplementary check across several
+# servers at once — a single slow fallback shouldn't hold up the others.
 _CORROBORATION_TIMEOUT_SECONDS = 10
 
 
@@ -223,27 +228,49 @@ class ElectrumListener:
         self.tip_height = height
         self.tip_header_hex = header_hex
 
-    async def corroborate_header(self, height: int, expected_hash: str) -> bool:
-        """B-28: independently ask every *other* configured server for the header
-        at `height` and require a strict majority of the ones that actually answer
-        to agree with `expected_hash` — the hash our own active connection
-        reported — before the draw (rounds/scheduler.py:_wait_for_next_block) treats
-        it as trustworthy entropy. Without this, a single hostile server (or a MITM
-        on the one active connection) can single-handedly decide who wins every
-        round; this raises the bar to controlling a majority of the configured
-        servers.
+    async def _corroborate_majority(
+        self,
+        ask: Callable[[ElectrumEndpoint], "asyncio.Future"],
+        agrees: Callable[[object], bool],
+        description: str,
+    ) -> bool:
+        """Shared quorum logic behind corroborate_header (B-28) and
+        corroborate_utxo_spent (B-29): ask every *other* configured server (never
+        the currently active one — that's exactly what a hostile server or a MITM
+        would control) and require a strict majority of the ones that actually
+        answer to agree, via `agrees`, with what our own connection reported.
 
-        Returns True if there are no other servers configured at all — a
-        single-endpoint deployment has nothing to corroborate against, and accepted
-        that risk when ELECTRUM_FALLBACK_SERVERS was left empty (see CLAUDE.md).
-        Also returns False (never silently "passes") if none of the other servers
-        could be reached at all, since an unreachable network answers nothing about
-        whether the header is genuine.
+        Returns True with no other servers configured — nothing to corroborate
+        against, a risk accepted when ELECTRUM_FALLBACK_SERVERS was left empty
+        (see CLAUDE.md). Returns False (never silently "passes") if none of the
+        others could be reached, since an unreachable network proves nothing
+        either way.
         """
         others = [endpoint for endpoint in self._endpoints if endpoint != self.current_endpoint]
         if not others:
             return True
 
+        results = await asyncio.gather(*(ask(endpoint) for endpoint in others))
+        responded = [result for result in results if result is not None]
+        if not responded:
+            logger.warning(
+                "could not corroborate %s with any of %s other configured server(s)", description, len(others)
+            )
+            return False
+
+        agreements = sum(1 for result in responded if agrees(result))
+        return agreements * 2 > len(responded)
+
+    async def corroborate_header(self, height: int, expected_hash: str) -> bool:
+        """B-28: is `expected_hash` — the header our own active connection
+        reported for `height` — corroborated by other configured servers before
+        the draw (rounds/scheduler.py:_wait_for_next_block) treats it as
+        trustworthy entropy? Without this, a single hostile server (or a MITM on
+        the one active connection) can single-handedly decide who wins every
+        round; this raises the bar to controlling a majority of the configured
+        servers. See _corroborate_majority for the shared quorum logic.
+        """
+
         async def _ask(endpoint: ElectrumEndpoint) -> str | None:
             client = self._client_factory(endpoint)
             try:
@@ -260,18 +287,32 @@ class ElectrumListener:
             finally:
                 await client.close()
 
-        results = await asyncio.gather(*(_ask(endpoint) for endpoint in others))
-        responded = [block_hash for block_hash in results if block_hash is not None]
-        if not responded:
-            logger.warning(
-                "could not corroborate block %s header with any of %s other configured server(s)",
-                height,
-                len(others),
-            )
-            return False
+        return await self._corroborate_majority(_ask, lambda block_hash: block_hash == expected_hash, f"block {height} header")
 
-        agreements = sum(1 for block_hash in responded if block_hash == expected_hash)
-        return agreements * 2 > len(responded)
+    async def corroborate_utxo_spent(self, scripthash: str, txid: str, vout: int) -> bool:
+        """B-29: before deposits/service.py's find_utxos_missing_from candidates
+        are treated as genuinely spent outside the platform, ask the other
+        configured servers whether *they* also no longer report this outpoint as
+        unspent. A single broken, behind, or malicious server could otherwise zero
+        a user's balance on one incomplete listunspent reply. See
+        _corroborate_majority for the shared quorum logic.
+        """
+
+        async def _ask(endpoint: ElectrumEndpoint) -> bool | None:
+            client = self._client_factory(endpoint)
+            try:
+                await asyncio.wait_for(client.connect(), timeout=_CORROBORATION_TIMEOUT_SECONDS)
+                entries = await asyncio.wait_for(
+                    client.listunspent(scripthash), timeout=_CORROBORATION_TIMEOUT_SECONDS
+                )
+                still_unspent = any(e.get("tx_hash") == txid and e.get("tx_pos") == vout for e in entries)
+                return not still_unspent  # True = this server agrees the outpoint is gone
+            except Exception:
+                return None
+            finally:
+                await client.close()
+
+        return await self._corroborate_majority(_ask, lambda agrees: agrees, f"outpoint {txid}:{vout}")
 
     async def _consume_headers(self, queue: asyncio.Queue) -> None:
         while True:
@@ -291,12 +332,35 @@ class ElectrumListener:
                 await self._refresh_user(user_id, scripthash)
 
     async def _refresh_user(self, user_id: int, scripthash: str) -> None:
+        """Three phases, so no DB session is held across a network call (B-18),
+        same shape as _trigger_payout: read what's needed, corroborate any
+        candidate external spends against other servers (B-29), then persist.
+        """
         assert self.client is not None
         entries = await self.client.listunspent(scripthash)
+
         async with self._session_factory() as session:
             credited = await credit_confirmed_utxos(session, user_id, entries)
-            spent_externally = await detect_external_spends(session, user_id, entries)
+            reinstated = await reinstate_reappeared_utxos(session, user_id, entries)
+            candidates = [
+                (row.id, row.txid, row.vout)
+                for row in await find_utxos_missing_from(session, user_id, entries)
+            ]
+
+        confirmed_ids = [
+            utxo_id
+            for utxo_id, txid, vout in candidates
+            if await self.corroborate_utxo_spent(scripthash, txid, vout)
+        ]
+
+        spent_externally = 0
+        if confirmed_ids:
+            async with self._session_factory() as session:
+                spent_externally = await mark_utxos_spent_externally(session, user_id, confirmed_ids)
+
         if credited:
             logger.info("credited %s new UTXO(s) for user_id=%s", credited, user_id)
+        if reinstated:
+            logger.info("reinstated %s previously-flagged UTXO(s) for user_id=%s", reinstated, user_id)
         if spent_externally:
             logger.warning("%s UTXO(s) spent outside the platform for user_id=%s", spent_externally, user_id)