Resubscribe concurrently and in the background on reconnect (B-31)

_run_once awaited _subscribe_all_users() inline, before starting the
header/scripthash consumer tasks, and that method subscribed one user
at a time. At thousands of users that's thousands of sequential
round-trips during which nothing else ran: tip_height was frozen and
an in-flight draw's _wait_for_next_block made zero progress for the
entire resubscribe - a reconnect (which the listener already treats as
routine, not exceptional) could stall the lottery for minutes.

_subscribe_all_users now fans out with bounded concurrency
(asyncio.Semaphore, 20 at a time) instead of a serial loop, and one
user's failure no longer stops the rest. _run_once now starts it as
its own background task, created after the consumer tasks rather than
awaited before them, so tip updates and already-subscribed users'
notifications keep flowing throughout - its own completion is
deliberately not raced against the session-ending tasks (unlike them,
it's expected to finish normally), and its failure is logged the same
way address_for_new_user's background task is (B-30).

Left out: decoupling the listunspent refresh from the subscribe call
itself (the third part of the proposed fix) - the periodic
DepositReconciler (B-30) already provides a backstop for a slow or
delayed initial refresh, so the added complexity wasn't worth it here.

Suite grows from 182 to 185 tests, including an end-to-end test
against _run_once proving a new tip is processed while a slow
resubscribe is still in flight. BUGS.md moves B-31 to Previously
fixed.

BUGS.md

@@ -1,11 +1,11 @@
 # Known bugs
 
 A second full-codebase audit on 2026-07-27 found **25 further issues** (4 critical, 6 high,
-7 medium, 8 low), listed below as B-31 … B-49. B-25 through B-30 are fixed (see "Previously
+7 medium, 8 low), listed below as B-32 … B-49. B-25 through B-31 are fixed (see "Previously
-fixed" below) — no Critical-severity finding remains open; the other 19 are High/Medium/Low.
+fixed" below) — no Critical-severity finding remains open; the other 18 are High/Medium/Low.
 The 139-test suite was green at the time of the audit, so none of these were caught by existing
-coverage — every fix lands with a regression test (the six fixes so far brought the suite from
+coverage — every fix lands with a regression test (the seven fixes so far brought the suite
-139 to 182).
+from 139 to 185).
 
 The recurring pattern across the open findings is worth stating once: the code is rigorous
 about the failure modes that have actually been hit, and silent about the ones that have not.
@@ -20,24 +20,6 @@ single-process assumptions, no user-facing history, etc.), see "Known gaps / TOD
 
 ## High
 
-### B-31 — Reconnect costs O(users) sequential round-trips and stalls the draw
-
-In `_run_once` the order is: subscribe headers → `_subscribe_all_users()` → *then* start the
-consumer tasks (`electrum/listener.py:118-134`). `_subscribe_all_users` iterates users
-**sequentially**, and each iteration is a subscribe plus a `listunspent` plus a DB write
-(`:154-165`).
-
-At 5.000 users that is 10.000 serialized round-trips (15s timeout each). Throughout,
-`_consume_headers` is not running, so `tip_height` is frozen and `_wait_for_next_block` makes
-no progress: **a reconnect stalls an in-flight draw** for the entire resubscribe. And since
-registration has no rate limiting, the user count is attacker-controlled.
-
-**Proposed fix.** Start the consumer tasks (headers especially) *before* resubscribing, so tip
-updates keep flowing during the sweep. Batch the resubscribe with bounded concurrency
-(e.g. `asyncio.Semaphore(20)` over `asyncio.gather`) instead of a serial loop, and decouple
-the `listunspent` refresh from the subscribe so the initial refresh can proceed in the
-background.
-
 ### B-32 — `bump_fee` can loop forever on rebroadcasts the node always rejects
 
 `tx/broadcast.py:87-88` forces `fee_delta = 1` when `fee_delta <= 0`. A **one-satoshi** total
@@ -265,9 +247,10 @@ already does.
 - **B-28** — a hostile Electrum server (or a MITM) could single-handedly pick the round's winner
 - **B-29** — a UTXO absent from one server's `listunspent` was marked spent immediately, irreversibly, on a single unauthenticated reply
 - **B-30** — a lost scripthash subscription meant a user's deposits were never credited, with no periodic safety net
+- **B-31** — resubscribing on reconnect ran serially before anything else started, freezing the chain tip (and so an in-flight draw) for the whole sweep
 
 See git history for the fix-by-fix breakdown (commits `f13f685`, `50a43ae`, `933760e`, and the
-B-28/B-29/B-30 fixes). Suite grew from 139 to 182 tests over the six.
+B-28/B-29/B-30/B-31 fixes). Suite grew from 139 to 185 tests over the seven.
 
 A full-codebase audit on 2026-07-26 (commit `d4e0974`) found 24 bugs across every Python
 module under `app/`, both static frontends, and the Docker/Caddy deployment — 5 critical,

app/electrum/listener.py

@@ -35,6 +35,11 @@ _PING_INTERVAL_SECONDS = 60
 # servers at once — a single slow fallback shouldn't hold up the others.
 _CORROBORATION_TIMEOUT_SECONDS = 10
 
+# How many users to resubscribe at once on reconnect (B-31), instead of one at a
+# time. Bounded rather than unlimited so a huge user base doesn't open thousands
+# of simultaneous in-flight requests against the one active connection.
+_RESUBSCRIBE_CONCURRENCY = 20
+
 
 class ElectrumListener:
     """Long-lived background task: keeps one Electrum connection open, subscribes
@@ -161,8 +166,6 @@ class ElectrumListener:
             header = await client.subscribe_headers()
             self._apply_header(header)
 
-            await self._subscribe_all_users()
-
             headers_queue = client.notifications("blockchain.headers.subscribe")
             scripthash_queue = client.notifications("blockchain.scripthash.subscribe")
             # The consumers below block on their queues forever by design, so they
@@ -175,9 +178,25 @@ class ElectrumListener:
                 asyncio.create_task(self._keepalive(client)),
                 asyncio.create_task(client.wait_closed()),
             ]
+            # B-31: resubscribing every user is O(users) sequential round-trips —
+            # at thousands of users that's minutes during which, previously,
+            # nothing above had started yet: tip_height was frozen and an
+            # in-flight draw's _wait_for_next_block made zero progress for the
+            # entire resubscribe. Running it as its own background task instead
+            # of awaiting it inline here means tip updates (and notifications for
+            # whichever users are already subscribed) keep flowing throughout.
+            # It's deliberately not one of the raced `tasks` above: unlike those,
+            # it's expected to finish normally, and its own completion must not
+            # look like the session ending. Any failure partway through is
+            # logged the same way address_for_new_user's background task is
+            # (B-30), and it's cancelled below along with everything else once
+            # the session actually does end.
+            subscribe_task = asyncio.create_task(self._subscribe_all_users())
+            subscribe_task.add_done_callback(self._log_subscribe_all_users_failure)
             try:
                 done, still_running = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
             finally:
+                subscribe_task.cancel()
                 for task in tasks:
                     task.cancel()
             for task in done:
@@ -189,18 +208,40 @@ class ElectrumListener:
             await client.close()
         return True
 
+    def _log_subscribe_all_users_failure(self, task: asyncio.Task) -> None:
+        if task.cancelled():
+            return
+        exc = task.exception()
+        if exc is not None:
+            logger.warning("resubscribing all users failed partway through: %r", exc)
+
     async def _keepalive(self, client: ElectrumClient) -> None:
         while True:
             await asyncio.sleep(_PING_INTERVAL_SECONDS)
             await client.ping()  # raises (and so ends the session) on timeout or a dead socket
 
     async def _subscribe_all_users(self) -> None:
+        """B-31: subscribes with bounded concurrency (_RESUBSCRIBE_CONCURRENCY at
+        a time) instead of one user at a time — at thousands of users a serial
+        loop meant thousands of sequential round-trips. One user's failure (a
+        single slow or briefly-erroring request) must not stop the rest from
+        being subscribed, mirroring the same per-item isolation used elsewhere
+        (e.g. tx/confirmation.py's poll_once, deposits/reconcile.py's sweep)."""
         async with self._session_factory() as session:
             users = (await session.scalars(select(User))).all()
-        for user in users:
+
+        semaphore = asyncio.Semaphore(_RESUBSCRIBE_CONCURRENCY)
+
+        async def _subscribe_one(user: User) -> None:
             scripthash = address_to_scripthash(user.address)
             self._scripthash_to_user[scripthash] = user.id
-            await self._subscribe_and_refresh(scripthash, user.id)
+            async with semaphore:
+                try:
+                    await self._subscribe_and_refresh(scripthash, user.id)
+                except Exception:
+                    logger.exception("failed to resubscribe user_id=%s", user.id)
+
+        await asyncio.gather(*(_subscribe_one(user) for user in users))
 
     async def _subscribe_and_refresh(self, scripthash: str, user_id: int) -> None:
         assert self.client is not None

tests/unit/test_electrum_listener.py

@@ -1,7 +1,8 @@
 """Listener-level behaviour: server rotation on failure (the fallback-servers
 feature), the chain-tip monotonicity guard (B-19), header validation and
-multi-server corroboration (B-28), and the new-user subscribe task's retention
+multi-server corroboration (B-28), the new-user subscribe task's retention and
-and error logging (B-30).
+error logging (B-30), and bounded-concurrency, non-blocking resubscribe on
+reconnect (B-31).
 
 The reconnect loop itself (B-01) is covered from the client side in
 test_electrum_client.py — what's asserted here is that the listener *acts* on a
@@ -442,7 +443,7 @@ async def _seed_funded_user(session_factory, *, username: str, address: str) ->
 _UNRELATED_ENTRY = [{"tx_hash": "99" * 32, "tx_pos": 0, "height": 100, "value": 1_000_000}]
 
 
-async def testrefresh_user_marks_a_utxo_spent_once_others_corroborate_it(session_factory):
+async def test_refresh_user_marks_a_utxo_spent_once_others_corroborate_it(session_factory):
     user_id = await _seed_funded_user(session_factory, username="bob", address="plm1q0xcqpzrky6eff2g52qdye53xkk9jxkvraxkkwd")
 
     others_factory = await _listunspent_client_factory(
@@ -464,7 +465,7 @@ async def testrefresh_user_marks_a_utxo_spent_once_others_corroborate_it(session
         assert user.cached_balance_sats == 1_000_000
 
 
-async def testrefresh_user_does_not_mark_when_corroboration_fails(session_factory):
+async def test_refresh_user_does_not_mark_when_corroboration_fails(session_factory):
     """The single most important case: our own connection alone reporting the
     UTXO missing must not be enough — before B-29 this zeroed the balance on one
     bad reply."""
@@ -487,3 +488,145 @@ async def testrefresh_user_does_not_mark_when_corroboration_fails(session_factor
         user = await session.get(User, user_id)
         # Untouched, plus the unrelated entry credited alongside it.
         assert user.cached_balance_sats == 21_000_000
+
+
+# --- B-31: resubscribing on reconnect must be bounded-concurrency and must not
+# block tip updates (and so an in-flight draw) for its entire duration. -----------
+
+
+def _fake_address(i: int) -> str:
+    """A real, decodable PLM bech32 P2WPKH address (address_to_scripthash
+    actually parses it) — distinct per index, since User.address is unique."""
+    from embit import script
+
+    from app.wallet.plm_network import PLM_MAINNET
+
+    payload = (i + 1).to_bytes(20, "big")
+    return script.Script(b"\x00\x14" + payload).address(network=PLM_MAINNET)
+
+
+async def _seed_users(session_factory, count: int) -> None:
+    async with session_factory() as session:
+        for i in range(count):
+            session.add(
+                User(username=f"user{i}", password_hash="x", derivation_index=i, address=_fake_address(i))
+            )
+        await session.commit()
+
+
+async def test_subscribe_all_users_bounds_concurrency(session_factory):
+    """B-31: at thousands of users, subscribing one at a time meant thousands of
+    sequential round-trips. Concurrency must be bounded (not unlimited either —
+    a huge user base shouldn't open thousands of simultaneous requests)."""
+    user_count = 45
+    await _seed_users(session_factory, user_count)
+
+    listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS)
+    in_flight = 0
+    max_in_flight = 0
+    calls = []
+
+    async def fake_subscribe_and_refresh(scripthash, user_id):
+        nonlocal in_flight, max_in_flight
+        in_flight += 1
+        max_in_flight = max(max_in_flight, in_flight)
+        calls.append(user_id)
+        await asyncio.sleep(0)  # yield, so genuinely-concurrent calls interleave
+        in_flight -= 1
+
+    listener._subscribe_and_refresh = fake_subscribe_and_refresh
+
+    await listener._subscribe_all_users()
+
+    assert len(calls) == user_count
+    assert 1 < max_in_flight <= 20  # bounded, and actually concurrent (not serial)
+
+
+async def test_subscribe_all_users_continues_past_a_failing_user(session_factory):
+    await _seed_users(session_factory, 5)
+    listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS)
+    succeeded = []
+
+    async def flaky_subscribe_and_refresh(scripthash, user_id):
+        if user_id == 3:
+            raise ConnectionResetError("dropped mid-subscribe")
+        succeeded.append(user_id)
+
+    listener._subscribe_and_refresh = flaky_subscribe_and_refresh
+
+    await listener._subscribe_all_users()  # must not raise
+
+    assert succeeded == [1, 2, 4, 5]
+
+
+class _FakeConnectClient:
+    """A minimally-real ElectrumClient double: enough of connect/subscribe/notify/
+    ping/wait_closed/close to drive ElectrumListener._run_once end-to-end."""
+
+    def __init__(self, header: dict):
+        self._header = header
+        self._queues: dict[str, asyncio.Queue] = {}
+        self._closed = asyncio.Event()
+
+    async def connect(self):
+        pass
+
+    async def subscribe_headers(self):
+        return self._header
+
+    def notifications(self, method: str) -> asyncio.Queue:
+        return self._queues.setdefault(method, asyncio.Queue())
+
+    async def ping(self):
+        pass
+
+    async def wait_closed(self):
+        await self._closed.wait()
+
+    async def close(self):
+        self._closed.set()
+
+
+async def _wait_until(predicate, *, timeout: float = 2.0, interval: float = 0.01) -> None:
+    async def _poll():
+        while not predicate():
+            await asyncio.sleep(interval)
+
+    await asyncio.wait_for(_poll(), timeout=timeout)
+
+
+async def test_run_once_keeps_consuming_headers_while_resubscribing(session_factory):
+    """The core B-31 fix: before this, _subscribe_all_users ran to completion
+    *before* the header-consuming task even started, so a reconnect with many
+    users froze tip_height — and so _wait_for_next_block's draw wait — for the
+    entire resubscribe. It must now keep advancing while resubscribing is still
+    in flight."""
+    await _seed_users(session_factory, 3)
+
+    header_hex = _mine_header("00" * 32)
+    client = _FakeConnectClient({"height": 100, "hex": header_hex})
+    listener = ElectrumListener(lambda endpoint: client, session_factory, _ENDPOINTS)
+
+    subscribe_started = asyncio.Event()
+
+    async def blocked_subscribe_and_refresh(scripthash, user_id):
+        subscribe_started.set()
+        await asyncio.sleep(3600)  # simulates a slow sweep; cancelled on cleanup
+
+    listener._subscribe_and_refresh = blocked_subscribe_and_refresh
+
+    run_once_task = asyncio.create_task(listener._run_once(_ENDPOINTS[0]))
+    try:
+        await asyncio.wait_for(subscribe_started.wait(), timeout=2)
+
+        # Resubscribing is still stuck mid-flight — but a new tip must still be
+        # processed, proving the header consumer isn't blocked behind it.
+        headers_queue = client.notifications("blockchain.headers.subscribe")
+        next_header_hex = _mine_header(header_hex_to_block_hash(header_hex))
+        await headers_queue.put([{"height": 101, "hex": next_header_hex}])
+        await _wait_until(lambda: listener.tip_height == 101)
+
+        assert listener.tip_header_hex == next_header_hex
+    finally:
+        await client.close()
+        await run_once_task