diff --git a/BUGS.md b/BUGS.md index 1ab436e..33e289a 100644 --- a/BUGS.md +++ b/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 -fixed" below) — no Critical-severity finding remains open; the other 19 are High/Medium/Low. +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 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 -139 to 182). +coverage — every fix lands with a regression test (the seven fixes so far brought the suite +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, diff --git a/app/electrum/listener.py b/app/electrum/listener.py index 55267cd..dab78bc 100644 --- a/app/electrum/listener.py +++ b/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 diff --git a/tests/unit/test_electrum_listener.py b/tests/unit/test_electrum_listener.py index 12e9afc..2270254 100644 --- a/tests/unit/test_electrum_listener.py +++ b/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 -and error logging (B-30). +multi-server corroboration (B-28), the new-user subscribe task's retention and +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